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.

- 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

Closes #1652

Author: Lucas Messenger

src/strands/models/bedrock.py

@@ -21,7 +21,7 @@
 
 from .._exception_notes import add_exception_note
 from ..event_loop import streaming
-from ..tools import convert_pydantic_to_tool_spec
+from ..tools import convert_pydantic_to_json_schema, convert_pydantic_to_tool_spec
 from ..tools._tool_helpers import noop_tool
 from ..types.content import ContentBlock, Messages, SystemContentBlock
 from ..types.exceptions import (
@@ -98,6 +98,9 @@ class BedrockConfig(TypedDict, total=False):
                 Please check https://docs.aws.amazon.com/bedrock/latest/userguide/service-tiers-inference.html for
                 supported service tiers, models, and regions
             stop_sequences: List of sequences that will stop generation when encountered
+            structured_output_mode: Mode for structured output. "tool" (default) uses tool-based approach,
+                "native" uses Bedrock's outputConfig.textFormat for schema-constrained responses.
+                Native mode requires a model that supports structured output.
             streaming: Flag to enable/disable streaming. Defaults to True.
             temperature: Controls randomness in generation (higher = more random)
             top_p: Controls diversity via nucleus sampling (alternative to temperature)
@@ -123,6 +126,7 @@ class BedrockConfig(TypedDict, total=False):
         include_tool_result_status: Literal["auto"] | bool | None
         service_tier: str | None
         stop_sequences: list[str] | None
+        structured_output_mode: Literal["tool", "native"] | None
         streaming: bool | None
         temperature: float | None
         top_p: float | None
@@ -218,6 +222,7 @@ def _format_request(
         tool_specs: list[ToolSpec] | None = None,
         system_prompt_content: list[SystemContentBlock] | None = None,
         tool_choice: ToolChoice | None = None,
+        output_config: dict[str, Any] | None = None,
     ) -> dict[str, Any]:
         """Format a Bedrock converse stream request.
 
@@ -226,6 +231,7 @@ def _format_request(
             tool_specs: List of tool specifications to make available to the model.
             tool_choice: Selection strategy for tool invocation.
             system_prompt_content: System prompt content blocks to provide context to the model.
+            output_config: Output configuration for structured output (JSON schema).
 
         Returns:
             A Bedrock converse stream request.
@@ -251,6 +257,20 @@ def _format_request(
             "messages": self._format_bedrock_messages(messages),
             "system": system_blocks,
             **({"serviceTier": {"type": self.config["service_tier"]}} if self.config.get("service_tier") else {}),
+            **(
+                {
+                    "outputConfig": {
+                        "textFormat": {
+                            "type": "json_schema",
+                            "structure": {
+                                "jsonSchema": output_config,
+                            },
+                        },
+                    }
+                }
+                if output_config
+                else {}
+            ),
             **(
                 {
                     "toolConfig": {
@@ -747,6 +767,7 @@ async def stream(
         *,
         tool_choice: ToolChoice | None = None,
         system_prompt_content: list[SystemContentBlock] | None = None,
+        output_config: dict[str, Any] | None = None,
         **kwargs: Any,
     ) -> AsyncGenerator[StreamEvent, None]:
         """Stream conversation with the Bedrock model.
@@ -760,6 +781,7 @@ async def stream(
             system_prompt: System prompt to provide context to the model.
             tool_choice: Selection strategy for tool invocation.
             system_prompt_content: System prompt content blocks to provide context to the model.
+            output_config: Output configuration for structured output (JSON schema).
             **kwargs: Additional keyword arguments for future extensibility.
 
         Yields:
@@ -782,7 +804,9 @@ def callback(event: StreamEvent | None = None) -> None:
         if system_prompt and system_prompt_content is None:
             system_prompt_content = [{"text": system_prompt}]
 
-        thread = asyncio.to_thread(self._stream, callback, messages, tool_specs, system_prompt_content, tool_choice)
+        thread = asyncio.to_thread(
+            self._stream, callback, messages, tool_specs, system_prompt_content, tool_choice, output_config
+        )
         task = asyncio.create_task(thread)
 
         while True:
@@ -801,6 +825,7 @@ def _stream(
         tool_specs: list[ToolSpec] | None = None,
         system_prompt_content: list[SystemContentBlock] | None = None,
         tool_choice: ToolChoice | None = None,
+        output_config: dict[str, Any] | None = None,
     ) -> None:
         """Stream conversation with the Bedrock model.
 
@@ -813,14 +838,15 @@ def _stream(
             tool_specs: List of tool specifications to make available to the model.
             system_prompt_content: System prompt content blocks to provide context to the model.
             tool_choice: Selection strategy for tool invocation.
+            output_config: Output configuration for structured output (JSON schema).
 
         Raises:
             ContextWindowOverflowException: If the input exceeds the model's context window.
             ModelThrottledException: If the model service is throttling requests.
         """
         try:
             logger.debug("formatting request")
-            request = self._format_request(messages, tool_specs, system_prompt_content, tool_choice)
+            request = self._format_request(messages, tool_specs, system_prompt_content, tool_choice, output_config)
             logger.debug("request=<%s>", request)
 
             logger.debug("invoking model")
@@ -1032,6 +1058,10 @@ async def structured_output(
     ) -> AsyncGenerator[dict[str, T | Any], None]:
         """Get structured output from the model.
 
+        Supports two modes controlled by `structured_output_mode` config:
+        - "tool" (default): Converts the Pydantic model to a tool spec and forces tool use.
+        - "native": Uses Bedrock's outputConfig.textFormat with JSON schema for guaranteed schema compliance.
+
         Args:
             output_model: The output model to use for the agent.
             prompt: The prompt messages to use for the agent.
@@ -1041,6 +1071,21 @@ async def structured_output(
         Yields:
             Model events with the last being the structured output.
         """
+        if self.config.get("structured_output_mode") == "native":
+            async for event in self._structured_output_native(output_model, prompt, system_prompt, **kwargs):
+                yield event
+        else:
+            async for event in self._structured_output_tool(output_model, prompt, system_prompt, **kwargs):
+                yield event
+
+    async def _structured_output_tool(
+        self,
+        output_model: type[T],
+        prompt: Messages,
+        system_prompt: str | None = None,
+        **kwargs: Any,
+    ) -> AsyncGenerator[dict[str, T | Any], None]:
+        """Structured output using tool-based approach."""
         tool_spec = convert_pydantic_to_tool_spec(output_model)
 
         response = self.stream(
@@ -1073,6 +1118,40 @@ async def structured_output(
 
         yield {"output": output_model(**output_response)}
 
+    async def _structured_output_native(
+        self,
+        output_model: type[T],
+        prompt: Messages,
+        system_prompt: str | None = None,
+        **kwargs: Any,
+    ) -> AsyncGenerator[dict[str, T | Any], None]:
+        """Structured output using Bedrock's native outputConfig.textFormat."""
+        output_config = convert_pydantic_to_json_schema(output_model)
+
+        response = self.stream(
+            messages=prompt,
+            system_prompt=system_prompt,
+            output_config=output_config,
+            **kwargs,
+        )
+        async for event in streaming.process_stream(response):
+            yield event
+
+        _, messages, _, _ = event["stop"]
+
+        content = messages["content"]
+        text_content: str | None = None
+        for block in content:
+            if "text" in block:
+                text_content = block["text"]
+                break
+
+        if text_content is None:
+            raise ValueError("No text content found in the Bedrock response for native structured output.")
+
+        output_response = json.loads(text_content)
+        yield {"output": output_model(**output_response)}
+
     @staticmethod
     def _get_default_model_with_warning(region_name: str, model_config: BedrockConfig | None = None) -> str:
         """Get the default Bedrock modelId based on region.

src/strands/tools/__init__.py

@@ -4,7 +4,7 @@
 """
 
 from .decorator import tool
-from .structured_output import convert_pydantic_to_tool_spec
+from .structured_output import convert_pydantic_to_json_schema, convert_pydantic_to_tool_spec
 from .tool_provider import ToolProvider
 from .tools import InvalidToolUseNameException, PythonAgentTool, normalize_schema, normalize_tool_spec
 
@@ -14,6 +14,7 @@
     "InvalidToolUseNameException",
     "normalize_schema",
     "normalize_tool_spec",
+    "convert_pydantic_to_json_schema",
     "convert_pydantic_to_tool_spec",
     "ToolProvider",
 ]

src/strands/tools/structured_output/__init__.py

@@ -1,6 +1,6 @@
 """Structured output tools for the Strands Agents framework."""
 
 from ._structured_output_context import DEFAULT_STRUCTURED_OUTPUT_PROMPT
-from .structured_output_utils import convert_pydantic_to_tool_spec
+from .structured_output_utils import convert_pydantic_to_json_schema, convert_pydantic_to_tool_spec
 
-__all__ = ["convert_pydantic_to_tool_spec", "DEFAULT_STRUCTURED_OUTPUT_PROMPT"]
+__all__ = ["convert_pydantic_to_json_schema", "convert_pydantic_to_tool_spec", "DEFAULT_STRUCTURED_OUTPUT_PROMPT"]

src/strands/tools/structured_output/structured_output_utils.py

@@ -1,5 +1,6 @@
 """Tools for converting Pydantic models to Bedrock tools."""
 
+import json
 from typing import Any, Union
 
 from pydantic import BaseModel
@@ -257,48 +258,56 @@ def _process_nested_dict(d: dict[str, Any], defs: dict[str, Any]) -> dict[str, A
     return result
 
 
-def convert_pydantic_to_tool_spec(
+def _prepare_pydantic_schema(
     model: type[BaseModel],
     description: str | None = None,
-) -> ToolSpec:
-    """Converts a Pydantic model to a tool description for the Amazon Bedrock Converse API.
+) -> tuple[str, str, dict[str, Any]]:
+    """Shared pipeline for converting a Pydantic model to a flattened JSON schema.
 
-    Handles optional vs. required fields, resolves $refs, and uses docstrings.
+    Resolves $refs, expands nested properties, flattens the schema, and resolves the description.
 
     Args:
-        model: The Pydantic model class to convert
-        description: Optional description of the tool's purpose
+        model: The Pydantic model class to convert.
+        description: Optional description override.
 
     Returns:
-        ToolSpec: Dict containing the Bedrock tool specification
+        Tuple of (name, description, flattened_schema).
     """
     name = model.__name__
 
-    # Get the JSON schema
     input_schema = model.model_json_schema()
 
-    # Get model docstring for description if not provided
     model_description = description
     if not model_description and model.__doc__:
         model_description = model.__doc__.strip()
 
-    # Process all referenced models to ensure proper docstrings
-    # This step is important for gathering descriptions from referenced models
     _process_referenced_models(input_schema, model)
-
-    # Now, let's fully expand the nested models with all their properties
     _expand_nested_properties(input_schema, model)
 
-    # Flatten the schema
-    flattened_schema = _flatten_schema(input_schema)
+    return name, model_description or "", _flatten_schema(input_schema)
+
+
+def convert_pydantic_to_tool_spec(
+    model: type[BaseModel],
+    description: str | None = None,
+) -> ToolSpec:
+    """Converts a Pydantic model to a tool description for the Amazon Bedrock Converse API.
+
+    Handles optional vs. required fields, resolves $refs, and uses docstrings.
+
+    Args:
+        model: The Pydantic model class to convert
+        description: Optional description of the tool's purpose
 
-    final_schema = flattened_schema
+    Returns:
+        ToolSpec: Dict containing the Bedrock tool specification
+    """
+    name, model_description, flattened_schema = _prepare_pydantic_schema(model, description)
 
-    # Construct the tool specification
     return ToolSpec(
         name=name,
         description=model_description or f"{name} structured output tool",
-        inputSchema={"json": final_schema},
+        inputSchema={"json": flattened_schema},
     )
 
 
@@ -402,3 +411,52 @@ def _process_properties(schema_def: dict[str, Any], model: type[BaseModel]) -> N
             # Add field description if available and not already set
             if field and field.description and not prop_info.get("description"):
                 prop_info["description"] = field.description
+
+
+def _add_additional_properties_false(schema: dict[str, Any]) -> None:
+    """Recursively add additionalProperties: false to all object types in a JSON schema.
+
+    Bedrock's native structured output requires additionalProperties: false at every object level.
+    Mutates the schema in place.
+
+    Args:
+        schema: The JSON schema to process (modified in place).
+    """
+    schema_type = schema.get("type")
+    if schema_type == "object" or (isinstance(schema_type, list) and "object" in schema_type):
+        schema["additionalProperties"] = False
+
+    if "properties" in schema:
+        for value in schema["properties"].values():
+            if isinstance(value, dict):
+                _add_additional_properties_false(value)
+
+    if "items" in schema and isinstance(schema["items"], dict):
+        _add_additional_properties_false(schema["items"])
+
+
+def convert_pydantic_to_json_schema(
+    model: type[BaseModel],
+    description: str | None = None,
+) -> dict[str, Any]:
+    """Convert a Pydantic model to a JSON schema dict for Bedrock native structured output.
+
+    Returns a dict with "schema" (JSON string), "name", and "description" keys,
+    suitable for use in outputConfig.textFormat.structure.jsonSchema.
+
+    Args:
+        model: The Pydantic model class to convert.
+        description: Optional description override.
+
+    Returns:
+        Dict with "schema" (JSON string), "name", and "description".
+    """
+    name, model_description, flattened_schema = _prepare_pydantic_schema(model, description)
+
+    _add_additional_properties_false(flattened_schema)
+
+    return {
+        "schema": json.dumps(flattened_schema),
+        "name": name,
+        "description": model_description or f"{name} structured output",
+    }