feat: add optional outputSchema support for tool specifications

- Add outputSchema as optional field in ToolSpec TypedDict
- Implement filter_tool_specs() method in Model base class
- Filter tool specs centrally in streaming.py before passing to models
- Add supports_tool_output_schema config option (default: False)
- Update test fixtures to support new filtering method
- Add comprehensive tests for outputSchema filtering

This allows agents to better understand tool outputs for improved
planning and tool chaining, while maintaining full backward compatibility.
Model providers that don't support outputSchema are protected by
filtering it out unless explicitly enabled.

Closes #787

Author: Vamil Gandhi

examples/output_schema_example.py

@@ -0,0 +1,112 @@
+"""Example demonstrating the use of outputSchema in tool specifications.
+
+This example shows how to define tools with outputSchema and how to enable
+or disable outputSchema support for different model providers.
+"""
+
+from strands.types.tools import ToolSpec
+
+# Example 1: Tool with outputSchema defined
+tool_with_output_schema = ToolSpec(
+    name="get_weather",
+    description="Get the current weather for a location",
+    inputSchema={
+        "json": {
+            "type": "object",
+            "properties": {
+                "location": {"type": "string", "description": "City name"},
+                "unit": {"type": "string", "enum": ["celsius", "fahrenheit"], "default": "celsius"},
+            },
+            "required": ["location"],
+        }
+    },
+    outputSchema={
+        "json": {
+            "type": "object",
+            "properties": {
+                "temperature": {"type": "number"},
+                "unit": {"type": "string"},
+                "description": {"type": "string"},
+                "humidity": {"type": "number", "minimum": 0, "maximum": 100},
+            },
+            "required": ["temperature", "unit", "description"],
+        }
+    },
+)
+
+# Example 2: Tool without outputSchema (backward compatible)
+tool_without_output_schema = ToolSpec(
+    name="calculate",
+    description="Perform a calculation",
+    inputSchema={
+        "json": {
+            "type": "object",
+            "properties": {"expression": {"type": "string", "description": "Mathematical expression"}},
+            "required": ["expression"],
+        }
+    },
+)
+
+
+# Example 3: Using with model providers
+def example_with_openai():
+    """Example using OpenAI with outputSchema support."""
+    from strands.models.openai import OpenAIModel
+
+    # Enable outputSchema support for compatible models
+    model = OpenAIModel(
+        model_id="gpt-4",
+        supports_tool_output_schema=True,  # Enable outputSchema support
+    )
+
+    # The model will receive tool specs with outputSchema included
+    # This helps the model understand the expected output format
+    return model
+
+
+def example_with_bedrock():
+    """Example using Bedrock (doesn't support outputSchema)."""
+    from strands.models.bedrock import BedrockModel
+
+    # By default, outputSchema is filtered out for compatibility
+    model = BedrockModel(
+        model_id="anthropic.claude-3-sonnet-20240229-v1:0",
+        # supports_tool_output_schema=False is the default
+    )
+
+    # The model will receive tool specs without outputSchema
+    # to avoid validation errors from the Bedrock API
+    return model
+
+
+def example_agent_planning():
+    """Example showing how agents can use outputSchema for planning.
+
+    When agents have access to outputSchema, they can:
+    1. Better understand what each tool returns
+    2. Generate more accurate execution plans
+    3. Prepare appropriate parameter values for subsequent tool calls
+    4. Chain tools together more effectively
+    """
+    tools = [tool_with_output_schema, tool_without_output_schema]
+
+    # Agents can inspect outputSchema to understand tool capabilities
+    for tool in tools:
+        print(f"Tool: {tool['name']}")
+        print(f"  Input: {tool['inputSchema']}")
+        if "outputSchema" in tool:
+            print(f"  Output: {tool['outputSchema']}")
+        else:
+            print("  Output: Not specified")
+
+    return tools
+
+
+if __name__ == "__main__":
+    print("Tool with outputSchema:")
+    print(tool_with_output_schema)
+    print("\nTool without outputSchema:")
+    print(tool_without_output_schema)
+    print("\n" + "=" * 50)
+    print("\nAgent planning example:")
+    example_agent_planning()

src/strands/event_loop/streaming.py

@@ -337,7 +337,9 @@ async def stream_messages(
     logger.debug("model=<%s> | streaming messages", model)
 
     messages = remove_blank_messages_content_text(messages)
-    chunks = model.stream(messages, tool_specs if tool_specs else None, system_prompt)
+    # Filter outputschema spec based on model configuration until all models supports it.
+    filtered_tool_specs = model.filter_tool_specs(tool_specs) if tool_specs else None
+    chunks = model.stream(messages, filtered_tool_specs, system_prompt)
 
     async for event in process_stream(chunks):
         yield event

src/strands/models/model.py

@@ -2,7 +2,7 @@
 
 import abc
 import logging
-from typing import Any, AsyncGenerator, AsyncIterable, Optional, Type, TypeVar, Union
+from typing import Any, AsyncGenerator, AsyncIterable, Optional, Type, TypeVar, Union, cast
 
 from pydantic import BaseModel
 
@@ -22,6 +22,27 @@ class Model(abc.ABC):
     standardized way to configure and process requests for different AI model providers.
     """
 
+    def filter_tool_specs(self, tool_specs: Optional[list[ToolSpec]]) -> Optional[list[ToolSpec]]:
+        """Filter tool specifications based on model configuration.
+
+        By default, this removes the outputSchema field from tool specs unless
+        the model configuration explicitly enables it via `supports_tool_output_schema`.
+
+        Args:
+            tool_specs: List of tool specifications to filter.
+
+        Returns:
+            Filtered tool specifications safe for the model provider.
+        """
+        if not tool_specs:
+            return tool_specs
+
+        config = self.get_config()
+        if isinstance(config, dict) and config.get("supports_tool_output_schema", False):
+            return tool_specs
+
+        return cast(list[ToolSpec], [{k: v for k, v in spec.items() if k != "outputSchema"} for spec in tool_specs])
+
     @abc.abstractmethod
     # pragma: no cover
     def update_config(self, **model_config: Any) -> None:

src/strands/types/tools.py

@@ -20,18 +20,22 @@
 """Type alias for JSON Schema dictionaries."""
 
 
-class ToolSpec(TypedDict):
+class ToolSpec(TypedDict, total=False):
     """Specification for a tool that can be used by an agent.
 
     Attributes:
         description: A human-readable description of what the tool does.
         inputSchema: JSON Schema defining the expected input parameters.
         name: The unique name of the tool.
+        outputSchema: Optional JSON Schema defining the expected output format.
+            Note: Not all model providers support this field. Providers that don't
+            support it should filter it out before sending to their API.
     """
 
     description: str
     inputSchema: JSONSchema
     name: str
+    outputSchema: JSONSchema
 
 
 class Tool(TypedDict):

tests/strands/agent/test_agent.py

@@ -49,6 +49,8 @@ async def stream(*args, **kwargs):
     mock = unittest.mock.Mock(spec=getattr(request, "param", None))
     mock.configure_mock(mock_stream=unittest.mock.MagicMock())
     mock.stream.side_effect = stream
+    mock.filter_tool_specs = lambda tool_specs: tool_specs
+    mock.get_config.return_value = {}
 
     return mock
 

tests/strands/event_loop/test_event_loop.py

@@ -33,7 +33,9 @@ def mock_time():
 
 @pytest.fixture
 def model():
-    return unittest.mock.Mock()
+    mock = unittest.mock.Mock()
+    mock.filter_tool_specs.side_effect = lambda tool_specs: tool_specs
+    return mock
 
 
 @pytest.fixture

tests/strands/models/test_output_schema_filtering.py

@@ -0,0 +1,153 @@
+"""Tests for outputSchema filtering in model providers."""
+
+from strands.models.model import Model
+from strands.types.tools import ToolSpec
+
+
+class MockModel(Model):
+    """Mock model implementation for testing."""
+
+    def __init__(self, supports_tool_output_schema: bool = False):
+        self.config = {"supports_tool_output_schema": supports_tool_output_schema}
+
+    def update_config(self, **model_config):
+        self.config.update(model_config)
+
+    def get_config(self):
+        return self.config
+
+    def structured_output(self, output_model, prompt, system_prompt=None, **kwargs):
+        pass
+
+    def stream(self, messages, tool_specs=None, system_prompt=None, **kwargs):
+        pass
+
+
+def test_filter_tool_specs_removes_output_schema_by_default():
+    """Test that outputSchema is removed when not explicitly supported."""
+    model = MockModel(supports_tool_output_schema=False)
+
+    tool_specs = [
+        ToolSpec(
+            name="test_tool",
+            description="A test tool",
+            inputSchema={"json": {"type": "object"}},
+            outputSchema={"json": {"type": "string"}},
+        )
+    ]
+
+    filtered = model.filter_tool_specs(tool_specs)
+
+    assert len(filtered) == 1
+    assert "name" in filtered[0]
+    assert "description" in filtered[0]
+    assert "inputSchema" in filtered[0]
+    assert "outputSchema" not in filtered[0]
+
+
+def test_filter_tool_specs_preserves_output_schema_when_supported():
+    """Test that outputSchema is preserved when explicitly supported."""
+    model = MockModel(supports_tool_output_schema=True)
+
+    tool_specs = [
+        ToolSpec(
+            name="test_tool",
+            description="A test tool",
+            inputSchema={"json": {"type": "object"}},
+            outputSchema={"json": {"type": "string"}},
+        )
+    ]
+
+    filtered = model.filter_tool_specs(tool_specs)
+
+    assert len(filtered) == 1
+    assert filtered[0] == tool_specs[0]
+    assert "outputSchema" in filtered[0]
+
+
+def test_filter_tool_specs_handles_missing_output_schema():
+    """Test that tools without outputSchema work correctly."""
+    model = MockModel(supports_tool_output_schema=False)
+
+    tool_specs = [ToolSpec(name="test_tool", description="A test tool", inputSchema={"json": {"type": "object"}})]
+
+    filtered = model.filter_tool_specs(tool_specs)
+
+    assert len(filtered) == 1
+    assert "name" in filtered[0]
+    assert "description" in filtered[0]
+    assert "inputSchema" in filtered[0]
+    assert "outputSchema" not in filtered[0]
+
+
+def test_filter_tool_specs_handles_none():
+    """Test that None tool_specs returns None."""
+    model = MockModel()
+
+    filtered = model.filter_tool_specs(None)
+
+    assert filtered is None
+
+
+def test_filter_tool_specs_handles_empty_list():
+    """Test that empty list returns empty list."""
+    model = MockModel()
+
+    filtered = model.filter_tool_specs([])
+
+    assert filtered == []
+
+
+def test_filter_tool_specs_multiple_tools():
+    """Test filtering multiple tools with mixed outputSchema presence."""
+    model = MockModel(supports_tool_output_schema=False)
+
+    tool_specs = [
+        ToolSpec(
+            name="tool1",
+            description="First tool",
+            inputSchema={"json": {"type": "object"}},
+            outputSchema={"json": {"type": "string"}},
+        ),
+        ToolSpec(name="tool2", description="Second tool", inputSchema={"json": {"type": "object"}}),
+        ToolSpec(
+            name="tool3",
+            description="Third tool",
+            inputSchema={"json": {"type": "object"}},
+            outputSchema={"json": {"type": "array"}},
+        ),
+    ]
+
+    filtered = model.filter_tool_specs(tool_specs)
+
+    assert len(filtered) == 3
+    for spec in filtered:
+        assert "outputSchema" not in spec
+        assert "name" in spec
+        assert "description" in spec
+        assert "inputSchema" in spec
+
+
+def test_update_config_changes_output_schema_support():
+    """Test that updating config can change outputSchema support."""
+    model = MockModel(supports_tool_output_schema=False)
+
+    tool_specs = [
+        ToolSpec(
+            name="test_tool",
+            description="A test tool",
+            inputSchema={"json": {"type": "object"}},
+            outputSchema={"json": {"type": "string"}},
+        )
+    ]
+
+    # Initially, outputSchema should be filtered out
+    filtered = model.filter_tool_specs(tool_specs)
+    assert "outputSchema" not in filtered[0]
+
+    # Update config to support outputSchema
+    model.update_config(supports_tool_output_schema=True)
+
+    # Now outputSchema should be preserved
+    filtered = model.filter_tool_specs(tool_specs)
+    assert "outputSchema" in filtered[0]