feat: Add Structured Output as part of the agent loop

src/strands/__init__.py

@@ -5,4 +5,12 @@
 from .tools.decorator import tool
 from .types.tools import ToolContext
 
-__all__ = ["Agent", "agent", "models", "tool", "types", "telemetry", "ToolContext"]
+__all__ = [
+    "Agent",
+    "agent",
+    "models",
+    "tool",
+    "ToolContext",
+    "types",
+    "telemetry",
+]

src/strands/agent/agent.py

@@ -29,6 +29,7 @@
 
 from opentelemetry import trace as trace_api
 from pydantic import BaseModel
+from typing_extensions import deprecated
 
 from .. import _identifier
 from ..event_loop.event_loop import event_loop_cycle
@@ -49,6 +50,7 @@
 from ..tools.executors import ConcurrentToolExecutor
 from ..tools.executors._executor import ToolExecutor
 from ..tools.registry import ToolRegistry
+from ..tools.structured_output.structured_output_context import StructuredOutputContext
 from ..tools.watcher import ToolWatcher
 from ..types._events import AgentResultEvent, InitEventLoopEvent, ModelStreamChunkEvent, TypedEvent
 from ..types.agent import AgentInput
@@ -210,6 +212,7 @@ def __init__(
         messages: Optional[Messages] = None,
         tools: Optional[list[Union[str, dict[str, str], Any]]] = None,
         system_prompt: Optional[str] = None,
+        structured_output_model: Optional[Type[BaseModel]] = None,
         callback_handler: Optional[
             Union[Callable[..., Any], _DefaultCallbackHandlerSentinel]
         ] = _DEFAULT_CALLBACK_HANDLER,
@@ -245,6 +248,10 @@ def __init__(
                 If provided, only these tools will be available. If None, all tools will be available.
             system_prompt: System prompt to guide model behavior.
                 If None, the model will behave according to its default settings.
+            structured_output_model: Pydantic model type(s) for structured output.
+                When specified, all agent calls will attempt to return structured output of this type.
+                This can be overridden on the agent invocation.
+                Defaults to None (no structured output).
             callback_handler: Callback for processing events as they happen during agent execution.
                 If not provided (using the default), a new PrintingCallbackHandler instance is created.
                 If explicitly set to None, null_callback_handler is used.
@@ -274,8 +281,8 @@ def __init__(
         """
         self.model = BedrockModel() if not model else BedrockModel(model_id=model) if isinstance(model, str) else model
         self.messages = messages if messages is not None else []
-
         self.system_prompt = system_prompt
+        self._default_structured_output_model = structured_output_model
         self.agent_id = _identifier.validate(agent_id or _DEFAULT_AGENT_ID, _identifier.Identifier.AGENT)
         self.name = name or _DEFAULT_AGENT_NAME
         self.description = description
@@ -374,7 +381,9 @@ def tool_names(self) -> list[str]:
         all_tools = self.tool_registry.get_all_tools_config()
         return list(all_tools.keys())
 
-    def __call__(self, prompt: AgentInput = None, **kwargs: Any) -> AgentResult:
+    def __call__(
+        self, prompt: AgentInput = None, structured_output_model: Type[BaseModel] | None = None, **kwargs: Any
+    ) -> AgentResult:
         """Process a natural language prompt through the agent's event loop.
 
         This method implements the conversational interface with multiple input patterns:
@@ -389,6 +398,7 @@ def __call__(self, prompt: AgentInput = None, **kwargs: Any) -> AgentResult:
                 - list[ContentBlock]: Multi-modal content blocks
                 - list[Message]: Complete messages with roles
                 - None: Use existing conversation history
+            structured_output_model: Pydantic model type(s) for structured output (overrides agent default).
             **kwargs: Additional parameters to pass through the event loop.
 
         Returns:
@@ -398,16 +408,19 @@ def __call__(self, prompt: AgentInput = None, **kwargs: Any) -> AgentResult:
                 - message: The final message from the model
                 - metrics: Performance metrics from the event loop
                 - state: The final state of the event loop
+                - structured_output: Parsed structured output when structured_output_model was specified
         """
 
         def execute() -> AgentResult:
-            return asyncio.run(self.invoke_async(prompt, **kwargs))
+            return asyncio.run(self.invoke_async(prompt, structured_output_model, **kwargs))
 
         with ThreadPoolExecutor() as executor:
             future = executor.submit(execute)
             return future.result()
 
-    async def invoke_async(self, prompt: AgentInput = None, **kwargs: Any) -> AgentResult:
+    async def invoke_async(
+        self, prompt: AgentInput = None, structured_output_model: Type[BaseModel] | None = None, **kwargs: Any
+    ) -> AgentResult:
         """Process a natural language prompt through the agent's event loop.
 
         This method implements the conversational interface with multiple input patterns:
@@ -422,6 +435,7 @@ async def invoke_async(self, prompt: AgentInput = None, **kwargs: Any) -> AgentR
                 - list[ContentBlock]: Multi-modal content blocks
                 - list[Message]: Complete messages with roles
                 - None: Use existing conversation history
+            structured_output_model: Pydantic model type(s) for structured output (overrides agent default).
             **kwargs: Additional parameters to pass through the event loop.
 
         Returns:
@@ -432,12 +446,17 @@ async def invoke_async(self, prompt: AgentInput = None, **kwargs: Any) -> AgentR
                 - metrics: Performance metrics from the event loop
                 - state: The final state of the event loop
         """
-        events = self.stream_async(prompt, **kwargs)
+        events = self.stream_async(prompt, structured_output_model=structured_output_model, **kwargs)
         async for event in events:
             _ = event
 
         return cast(AgentResult, event["result"])
 
+    @deprecated(
+        "Agent.structured_output method is deprecated."
+        " You should pass in `structured_output_model` directly into the agent invocation."
+        " see the <LINK> for more details"
+    )
     def structured_output(self, output_model: Type[T], prompt: AgentInput = None) -> T:
         """This method allows you to get structured output from the agent.
 
@@ -467,6 +486,11 @@ def execute() -> T:
             future = executor.submit(execute)
             return future.result()
 
+    @deprecated(
+        "Agent.structured_output_async method is deprecated."
+        " You should pass in `structured_output_model` directly into the agent invocation."
+        " see the <LINK> for more details"
+    )
     async def structured_output_async(self, output_model: Type[T], prompt: AgentInput = None) -> T:
         """This method allows you to get structured output from the agent.
 
@@ -530,6 +554,7 @@ async def structured_output_async(self, output_model: Type[T], prompt: AgentInpu
     async def stream_async(
         self,
         prompt: AgentInput = None,
+        structured_output_model: Type[BaseModel] | None = None,
         **kwargs: Any,
     ) -> AsyncIterator[Any]:
         """Process a natural language prompt and yield events as an async iterator.
@@ -546,6 +571,7 @@ async def stream_async(
                 - list[ContentBlock]: Multi-modal content blocks
                 - list[Message]: Complete messages with roles
                 - None: Use existing conversation history
+            structured_output_model: Pydantic model type(s) for structured output (overrides agent default).
             **kwargs: Additional parameters to pass to the event loop.
 
         Yields:
@@ -576,7 +602,7 @@ async def stream_async(
 
         with trace_api.use_span(self.trace_span):
             try:
-                events = self._run_loop(messages, invocation_state=kwargs)
+                events = self._run_loop(messages, kwargs, structured_output_model)
 
                 async for event in events:
                     event.prepare(invocation_state=kwargs)
@@ -596,12 +622,18 @@ async def stream_async(
                 self._end_agent_trace_span(error=e)
                 raise
 
-    async def _run_loop(self, messages: Messages, invocation_state: dict[str, Any]) -> AsyncGenerator[TypedEvent, None]:
+    async def _run_loop(
+        self,
+        messages: Messages,
+        invocation_state: dict[str, Any],
+        structured_output_model: Type[BaseModel] | None = None,
+    ) -> AsyncGenerator[TypedEvent, None]:
         """Execute the agent's event loop with the given message and parameters.
 
         Args:
             messages: The input messages to add to the conversation.
             invocation_state: Additional parameters to pass to the event loop.
+            structured_output_model: Optional Pydantic model type for structured output.
 
         Yields:
             Events from the event loop cycle.
@@ -614,8 +646,12 @@ async def _run_loop(self, messages: Messages, invocation_state: dict[str, Any])
             for message in messages:
                 self._append_message(message)
 
+            structured_output_context = StructuredOutputContext(
+                structured_output_model or self._default_structured_output_model
+            )
+
             # Execute the event loop cycle with retry logic for context limits
-            events = self._execute_event_loop_cycle(invocation_state)
+            events = self._execute_event_loop_cycle(invocation_state, structured_output_context)
             async for event in events:
                 # Signal from the model provider that the message sent by the user should be redacted,
                 # likely due to a guardrail.
@@ -636,24 +672,33 @@ async def _run_loop(self, messages: Messages, invocation_state: dict[str, Any])
             self.conversation_manager.apply_management(self)
             self.hooks.invoke_callbacks(AfterInvocationEvent(agent=self))
 
-    async def _execute_event_loop_cycle(self, invocation_state: dict[str, Any]) -> AsyncGenerator[TypedEvent, None]:
+    async def _execute_event_loop_cycle(
+        self, invocation_state: dict[str, Any], structured_output_context: StructuredOutputContext | None = None
+    ) -> AsyncGenerator[TypedEvent, None]:
         """Execute the event loop cycle with retry logic for context window limits.
 
         This internal method handles the execution of the event loop cycle and implements
         retry logic for handling context window overflow exceptions by reducing the
         conversation context and retrying.
 
+        Args:
+            invocation_state: Additional parameters to pass to the event loop.
+            structured_output_context: Optional structured output context for this invocation.
+
         Yields:
             Events of the loop cycle.
         """
         # Add `Agent` to invocation_state to keep backwards-compatibility
         invocation_state["agent"] = self
 
+        if structured_output_context and structured_output_context.structured_output_tool:
+            self.tool_registry.register_dynamic_tool(structured_output_context.structured_output_tool)
+
         try:
-            # Execute the main event loop cycle
             events = event_loop_cycle(
                 agent=self,
                 invocation_state=invocation_state,
+                structured_output_context=structured_output_context,
             )
             async for event in events:
                 yield event
@@ -666,7 +711,7 @@ async def _execute_event_loop_cycle(self, invocation_state: dict[str, Any]) -> A
             if self._session_manager:
                 self._session_manager.sync_agent(self)
 
-            events = self._execute_event_loop_cycle(invocation_state)
+            events = self._execute_event_loop_cycle(invocation_state, structured_output_context)
             async for event in events:
                 yield event
 

src/strands/agent/agent_result.py

@@ -6,6 +6,8 @@
 from dataclasses import dataclass
 from typing import Any
 
+from pydantic import BaseModel
+
 from ..telemetry.metrics import EventLoopMetrics
 from ..types.content import Message
 from ..types.streaming import StopReason
@@ -20,12 +22,14 @@ class AgentResult:
         message: The last message generated by the agent.
         metrics: Performance metrics collected during processing.
         state: Additional state information from the event loop.
+        structured_output: Parsed structured output when structured_output_model was specified.
     """
 
     stop_reason: StopReason
     message: Message
     metrics: EventLoopMetrics
     state: Any
+    structured_output: BaseModel | None = None
 
     def __str__(self) -> str:
         """Get the agent's last message as a string.