feat(bedrock): add native structured output support via outputConfig.textFormat

Add opt-in native structured output mode for BedrockModel that uses
Bedrock's outputConfig.textFormat API for schema-constrained responses,
replacing the tool-based workaround when enabled.

Model-level:
- Add `structured_output_mode` config ("tool" | "native", defaults to "tool")
- Add `convert_pydantic_to_json_schema()` utility with recursive
  `additionalProperties: false` injection
- Thread `output_config` through stream() -> _stream() -> _format_request()
- Native mode parses JSON text response instead of extracting tool use args

Agent-level:
- StructuredOutputContext gains native_mode: skips tool registration,
  stores output_config, and extracts results from text responses
- event_loop_cycle handles end_turn as success in native mode
- stream_messages passes output_config through to model.stream()

Closes #1652

Author: Lucas Messenger

examples/bedrock_native_structured_output.py

@@ -0,0 +1,111 @@
+"""Example: Bedrock native structured output using outputConfig.textFormat.
+
+Demonstrates three ways to use native structured output:
+1. Agent-level: agent(prompt, structured_output_model=Model)
+2. Model-level via agent: agent.structured_output(Model, prompt)
+3. Model-level direct: model.structured_output(Model, prompt)
+
+Requires a Bedrock model that supports structured output (e.g., Claude 4.5+).
+"""
+
+import asyncio
+import logging
+from pprint import pprint
+
+from pydantic import BaseModel, Field
+
+from strands import Agent
+from strands.models import BedrockModel
+
+# Enable debug logging to confirm native mode (no toolConfig, has outputConfig)
+logging.basicConfig(level=logging.WARNING, format="%(name)s %(levelname)s %(message)s")
+logging.getLogger("strands.models.bedrock").setLevel(logging.DEBUG)
+logging.getLogger("strands.event_loop.event_loop").setLevel(logging.DEBUG)
+
+
+# --- Models ---
+
+
+class WeatherReport(BaseModel):
+    """A weather report for a given location."""
+
+    location: str = Field(description="The city and state")
+    temperature_f: int = Field(description="Temperature in Fahrenheit")
+    conditions: str = Field(description="Current weather conditions")
+    wind_mph: int = Field(description="Wind speed in miles per hour")
+
+
+class MovieReview(BaseModel):
+    """A movie review with rating."""
+
+    title: str = Field(description="The movie title")
+    year: int = Field(description="Release year")
+    rating: float = Field(description="Rating out of 10")
+    summary: str = Field(description="One sentence summary")
+    pros: list[str] = Field(description="List of positive aspects")
+    cons: list[str] = Field(description="List of negative aspects")
+
+
+class Recipe(BaseModel):
+    """A cooking recipe."""
+
+    name: str = Field(description="Recipe name")
+    servings: int = Field(description="Number of servings")
+    prep_time_minutes: int = Field(description="Preparation time in minutes")
+    ingredients: list[str] = Field(description="List of ingredients")
+    steps: list[str] = Field(description="Ordered cooking steps")
+
+
+# --- Setup ---
+
+model = BedrockModel(
+    model_id="us.anthropic.claude-sonnet-4-5-20250929-v1:0",
+    structured_output_mode="native",
+    additional_request_fields={"thinking": {"type": "enabled", "budget_tokens": 1024}},
+)
+agent = Agent(model=model)
+
+
+# --- Example 1: Agent-level structured output ---
+
+print("=" * 60)
+print("Example 1: Agent-level (recommended)")
+print("=" * 60)
+
+result = agent(
+    "What's the weather like in Seattle, WA right now? Make up plausible data.",
+    structured_output_model=WeatherReport,
+)
+weather = result.structured_output
+pprint(weather)
+
+
+# --- Example 2: Model-level via agent.structured_output ---
+
+print("=" * 60)
+print("Example 2: Model-level via agent.structured_output()")
+print("=" * 60)
+
+review = agent.structured_output(MovieReview, "Write a review of The Matrix (1999)")
+pprint(review)
+
+
+# --- Example 3: Direct model.structured_output ---
+
+print("=" * 60)
+print("Example 3: Direct model.structured_output()")
+print("=" * 60)
+
+
+async def direct_model_example():
+    messages = [{"role": "user", "content": [{"text": "Give me a simple pancake recipe"}]}]
+    stream = model.structured_output(Recipe, messages)
+    result = None
+    async for event in stream:
+        if "output" in event:
+            result = event["output"]
+    return result
+
+
+recipe = asyncio.run(direct_model_example())
+pprint(recipe)

src/strands/agent/agent.py

@@ -886,9 +886,14 @@ async def _run_loop(
 
                 await self._append_messages(*current_messages)
 
+                # Check if the model supports native structured output
+                model_config = self.model.get_config()
+                native_mode = isinstance(model_config, dict) and model_config.get("structured_output_mode") == "native"
+
                 structured_output_context = StructuredOutputContext(
                     structured_output_model or self._default_structured_output_model,
                     structured_output_prompt=structured_output_prompt or self._structured_output_prompt,
+                    native_mode=native_mode,
                 )
 
                 # Execute the event loop cycle with retry logic for context limits
@@ -950,7 +955,7 @@ async def _execute_event_loop_cycle(
         # Add `Agent` to invocation_state to keep backwards-compatibility
         invocation_state["agent"] = self
 
-        if structured_output_context:
+        if structured_output_context and not structured_output_context.native_mode:
             structured_output_context.register_tool(self.tool_registry)
 
         try:

src/strands/event_loop/event_loop.py

@@ -201,25 +201,46 @@ async def event_loop_cycle(
             # End the cycle and return results
             agent.event_loop_metrics.end_cycle(cycle_start_time, cycle_trace, attributes)
 
-            # Force structured output tool call if LLM didn't use it automatically
+            # Handle structured output when model returns end_turn
             if structured_output_context.is_enabled and stop_reason == "end_turn":
-                if structured_output_context.force_attempted:
+                if structured_output_context.native_mode:
+                    # Native mode: parse text response as structured output
+                    native_result = structured_output_context.extract_native_result(message)
+                    if native_result is not None:
+                        yield StructuredOutputEvent(structured_output=native_result)
+                        tracer.end_event_loop_cycle_span(cycle_span, message)
+                        yield EventLoopStopEvent(
+                            stop_reason,
+                            message,
+                            agent.event_loop_metrics,
+                            invocation_state["request_state"],
+                            structured_output=native_result,
+                        )
+                        return
                     raise StructuredOutputException(
-                        "The model failed to invoke the structured output tool even after it was forced."
+                        "Native structured output mode: no text content found in model response."
+                    )
+                else:
+                    # Tool mode: force the model to call the structured output tool
+                    if structured_output_context.force_attempted:
+                        raise StructuredOutputException(
+                            "The model failed to invoke the structured output tool even after it was forced."
+                        )
+                    structured_output_context.set_forced_mode()
+                    logger.debug("Forcing structured output tool")
+                    await agent._append_messages(
+                        {"role": "user", "content": [{"text": structured_output_context.structured_output_prompt}]}
                     )
-                structured_output_context.set_forced_mode()
-                logger.debug("Forcing structured output tool")
-                await agent._append_messages(
-                    {"role": "user", "content": [{"text": structured_output_context.structured_output_prompt}]}
-                )
 
-                tracer.end_event_loop_cycle_span(cycle_span, message)
-                events = recurse_event_loop(
-                    agent=agent, invocation_state=invocation_state, structured_output_context=structured_output_context
-                )
-                async for typed_event in events:
-                    yield typed_event
-                return
+                    tracer.end_event_loop_cycle_span(cycle_span, message)
+                    events = recurse_event_loop(
+                        agent=agent,
+                        invocation_state=invocation_state,
+                        structured_output_context=structured_output_context,
+                    )
+                    async for typed_event in events:
+                        yield typed_event
+                    return
 
             tracer.end_event_loop_cycle_span(cycle_span, message)
             yield EventLoopStopEvent(stop_reason, message, agent.event_loop_metrics, invocation_state["request_state"])
@@ -338,6 +359,15 @@ async def _handle_model_execution(
                 else:
                     tool_specs = agent.tool_registry.get_all_tool_specs()
 
+                if structured_output_context.is_enabled:
+                    so_mode = "native" if structured_output_context.native_mode else "tool"
+                    logger.debug(
+                        "structured_output_mode=<%s> | tool_count=<%d> | output_config=<%s>",
+                        so_mode,
+                        len(tool_specs),
+                        bool(structured_output_context.output_config),
+                    )
+
                 async for event in stream_messages(
                     agent.model,
                     agent.system_prompt,
@@ -348,6 +378,7 @@ async def _handle_model_execution(
                     invocation_state=invocation_state,
                     model_state=agent._model_state,
                     cancel_signal=agent._cancel_signal,
+                    output_config=structured_output_context.output_config,
                 ):
                     yield event
 

src/strands/event_loop/streaming.py

@@ -490,6 +490,11 @@ async def stream_messages(
     messages = _normalize_messages(messages)
     start_time = time.time()
 
+    # Build optional kwargs for model.stream() - only pass output_config if supported
+    stream_kwargs: dict[str, Any] = {}
+    if kwargs.get("output_config") is not None:
+        stream_kwargs["output_config"] = kwargs["output_config"]
+
     chunks = model.stream(
         messages,
         tool_specs if tool_specs else None,
@@ -498,6 +503,7 @@ async def stream_messages(
         system_prompt_content=system_prompt_content,
         invocation_state=invocation_state,
         model_state=model_state,
+        **stream_kwargs,
     )
 
     async for event in process_stream(chunks, start_time, cancel_signal):