feat: add optional outputSchema support for tool specifications

src/strands/models/bedrock.py

@@ -214,7 +214,16 @@ def format_request(
                 {
                     "toolConfig": {
                         "tools": [
-                            *[{"toolSpec": tool_spec} for tool_spec in tool_specs],
+                            *[
+                                {
+                                    "toolSpec": {
+                                        "name": tool_spec["name"],
+                                        "description": tool_spec["description"],
+                                        "inputSchema": tool_spec["inputSchema"],
+                                    }
+                                }
+                                for tool_spec in tool_specs
+                            ],
                             *(
                                 [{"cachePoint": {"type": self.config["cache_tools"]}}]
                                 if self.config.get("cache_tools")

src/strands/tools/mcp/mcp_agent_tool.py

@@ -54,18 +54,24 @@ def tool_spec(self) -> ToolSpec:
         """Get the specification of the tool.
 
         This method converts the MCP tool specification to the agent framework's
-        ToolSpec format, including the input schema and description.
+        ToolSpec format, including the input schema, description, and optional output schema.
 
         Returns:
             ToolSpec: The tool specification in the agent framework format
         """
         description: str = self.mcp_tool.description or f"Tool which performs {self.mcp_tool.name}"
-        return {
+
+        spec: ToolSpec = {
             "inputSchema": {"json": self.mcp_tool.inputSchema},
             "name": self.mcp_tool.name,
             "description": description,
         }
 
+        if self.mcp_tool.outputSchema:
+            spec["outputSchema"] = {"json": self.mcp_tool.outputSchema}
+
+        return spec
+
     @property
     def tool_type(self) -> str:
         """Get the type of the tool.

src/strands/types/tools.py

@@ -9,7 +9,7 @@
 from dataclasses import dataclass
 from typing import TYPE_CHECKING, Any, AsyncGenerator, Awaitable, Callable, Literal, Protocol, Union
 
-from typing_extensions import TypedDict
+from typing_extensions import NotRequired, TypedDict
 
 from .media import DocumentContent, ImageContent
 
@@ -27,11 +27,15 @@ class ToolSpec(TypedDict):
         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: NotRequired[JSONSchema]
 
 
 class Tool(TypedDict):

tests/strands/models/test_bedrock.py

@@ -1788,3 +1788,25 @@ def test_custom_model_id_not_overridden_by_region_formatting(session_cls):
     model_id = model.get_config().get("model_id")
 
     assert model_id == custom_model_id
+
+
+def test_format_request_filters_output_schema(model, messages, model_id):
+    """Test that outputSchema is filtered out from tool specs in Bedrock requests."""
+    tool_spec_with_output_schema = {
+        "description": "Test tool with output schema",
+        "name": "test_tool",
+        "inputSchema": {"type": "object", "properties": {}},
+        "outputSchema": {"type": "object", "properties": {"result": {"type": "string"}}},
+    }
+
+    request = model.format_request(messages, [tool_spec_with_output_schema])
+
+    tool_spec = request["toolConfig"]["tools"][0]["toolSpec"]
+
+    # Verify outputSchema is not included
+    assert "outputSchema" not in tool_spec
+
+    # Verify other fields are preserved
+    assert tool_spec["name"] == "test_tool"
+    assert tool_spec["description"] == "Test tool with output schema"
+    assert tool_spec["inputSchema"] == {"type": "object", "properties": {}}

tests/strands/tools/mcp/test_mcp_agent_tool.py

@@ -13,6 +13,7 @@ def mock_mcp_tool():
     mock_tool.name = "test_tool"
     mock_tool.description = "A test tool"
     mock_tool.inputSchema = {"type": "object", "properties": {}}
+    mock_tool.outputSchema = None  # MCP tools can have optional outputSchema
     return mock_tool
 
 
@@ -47,6 +48,7 @@ def test_tool_spec_with_description(mcp_agent_tool, mock_mcp_tool):
     assert tool_spec["name"] == "test_tool"
     assert tool_spec["description"] == "A test tool"
     assert tool_spec["inputSchema"]["json"] == {"type": "object", "properties": {}}
+    assert "outputSchema" not in tool_spec
 
 
 def test_tool_spec_without_description(mock_mcp_tool, mock_mcp_client):
@@ -58,6 +60,25 @@ def test_tool_spec_without_description(mock_mcp_tool, mock_mcp_client):
     assert tool_spec["description"] == "Tool which performs test_tool"
 
 
+def test_tool_spec_with_output_schema(mock_mcp_tool, mock_mcp_client):
+    mock_mcp_tool.outputSchema = {"type": "object", "properties": {"result": {"type": "string"}}}
+
+    agent_tool = MCPAgentTool(mock_mcp_tool, mock_mcp_client)
+    tool_spec = agent_tool.tool_spec
+
+    assert "outputSchema" in tool_spec
+    assert tool_spec["outputSchema"]["json"] == {"type": "object", "properties": {"result": {"type": "string"}}}
+
+
+def test_tool_spec_without_output_schema(mock_mcp_tool, mock_mcp_client):
+    mock_mcp_tool.outputSchema = None
+
+    agent_tool = MCPAgentTool(mock_mcp_tool, mock_mcp_client)
+    tool_spec = agent_tool.tool_spec
+
+    assert "outputSchema" not in tool_spec
+
+
 @pytest.mark.asyncio
 async def test_stream(mcp_agent_tool, mock_mcp_client, alist):
     tool_use = {"toolUseId": "test-123", "name": "test_tool", "input": {"param": "value"}}

tests_integ/mcp/__init__.py

@@ -0,0 +1 @@
+"""MCP integration tests package."""

tests_integ/echo_server.py → tests_integ/mcp/echo_server.py

@@ -15,9 +15,15 @@
     $ python echo_server.py
 """
 
-from typing import Any, Dict
-
 from mcp.server import FastMCP
+from pydantic import BaseModel
+
+
+class EchoResponse(BaseModel):
+    """Response model for echo with structured content."""
+
+    echoed: str
+    message_length: int
 
 
 def start_echo_server():
@@ -37,8 +43,8 @@ def echo(to_echo: str) -> str:
 
     # FastMCP automatically constructs structured output schema from method signature
     @mcp.tool(description="Echos response back with structured content", structured_output=True)
-    def echo_with_structured_content(to_echo: str) -> Dict[str, Any]:
-        return {"echoed": to_echo}
+    def echo_with_structured_content(to_echo: str) -> EchoResponse:
+        return EchoResponse(echoed=to_echo, message_length=len(to_echo))
 
     mcp.run(transport="stdio")
 

tests_integ/test_mcp_client.py → tests_integ/mcp/test_mcp_client.py

@@ -76,7 +76,7 @@ def test_mcp_client():
 
     sse_mcp_client = MCPClient(lambda: sse_client("http://127.0.0.1:8000/sse"))
     stdio_mcp_client = MCPClient(
-        lambda: stdio_client(StdioServerParameters(command="python", args=["tests_integ/echo_server.py"]))
+        lambda: stdio_client(StdioServerParameters(command="python", args=["tests_integ/mcp/echo_server.py"]))
     )
 
     with sse_mcp_client, stdio_mcp_client:
@@ -150,19 +150,19 @@ def test_mcp_client():
 
         # With the new MCPToolResult, structured content is in its own field
         assert "structuredContent" in result
-        assert result["structuredContent"]["result"] == {"echoed": "STRUCTURED_DATA_TEST"}
+        assert result["structuredContent"] == {"echoed": "STRUCTURED_DATA_TEST", "message_length": 20}
 
         # Verify the result is an MCPToolResult (at runtime it's just a dict, but type-wise it should be MCPToolResult)
         assert result["status"] == "success"
         assert result["toolUseId"] == tool_use_id
 
         assert len(result["content"]) == 1
-        assert json.loads(result["content"][0]["text"]) == {"echoed": "STRUCTURED_DATA_TEST"}
+        assert json.loads(result["content"][0]["text"]) == {"echoed": "STRUCTURED_DATA_TEST", "message_length": 20}
 
 
 def test_can_reuse_mcp_client():
     stdio_mcp_client = MCPClient(
-        lambda: stdio_client(StdioServerParameters(command="python", args=["tests_integ/echo_server.py"]))
+        lambda: stdio_client(StdioServerParameters(command="python", args=["tests_integ/mcp/echo_server.py"]))
     )
     with stdio_mcp_client:
         stdio_mcp_client.list_tools_sync()
@@ -185,7 +185,7 @@ async def test_mcp_client_async_structured_content():
     that appears in structuredContent field.
     """
     stdio_mcp_client = MCPClient(
-        lambda: stdio_client(StdioServerParameters(command="python", args=["tests_integ/echo_server.py"]))
+        lambda: stdio_client(StdioServerParameters(command="python", args=["tests_integ/mcp/echo_server.py"]))
     )
 
     with stdio_mcp_client:
@@ -200,20 +200,20 @@ async def test_mcp_client_async_structured_content():
         assert "structuredContent" in result
         # "result" nesting is not part of the MCP Structured Content specification,
         # but rather a FastMCP implementation detail
-        assert result["structuredContent"]["result"] == {"echoed": "ASYNC_STRUCTURED_TEST"}
+        assert result["structuredContent"] == {"echoed": "ASYNC_STRUCTURED_TEST", "message_length": 21}
 
         # Verify basic MCPToolResult structure
         assert result["status"] in ["success", "error"]
         assert result["toolUseId"] == tool_use_id
 
         assert len(result["content"]) == 1
-        assert json.loads(result["content"][0]["text"]) == {"echoed": "ASYNC_STRUCTURED_TEST"}
+        assert json.loads(result["content"][0]["text"]) == {"echoed": "ASYNC_STRUCTURED_TEST", "message_length": 21}
 
 
 def test_mcp_client_without_structured_content():
     """Test that MCP client works correctly when tools don't return structured content."""
     stdio_mcp_client = MCPClient(
-        lambda: stdio_client(StdioServerParameters(command="python", args=["tests_integ/echo_server.py"]))
+        lambda: stdio_client(StdioServerParameters(command="python", args=["tests_integ/mcp/echo_server.py"]))
     )
 
     with stdio_mcp_client:
@@ -279,7 +279,7 @@ def test_mcp_client_timeout_integration():
 
     def slow_transport():
         time.sleep(4)  # Longer than timeout
-        return stdio_client(StdioServerParameters(command="python", args=["tests_integ/echo_server.py"]))
+        return stdio_client(StdioServerParameters(command="python", args=["tests_integ/mcp/echo_server.py"]))
 
     client = MCPClient(slow_transport, startup_timeout=2)
     initial_threads = threading.active_count()

tests_integ/test_mcp_client_structured_content_with_hooks.py → tests_integ/mcp/test_mcp_client_structured_content_with_hooks.py

@@ -37,7 +37,7 @@ def test_mcp_client_hooks_structured_content():
 
     # Set up MCP client for echo server
     stdio_mcp_client = MCPClient(
-        lambda: stdio_client(StdioServerParameters(command="python", args=["tests_integ/echo_server.py"]))
+        lambda: stdio_client(StdioServerParameters(command="python", args=["tests_integ/mcp/echo_server.py"]))
     )
 
     with stdio_mcp_client:
@@ -58,8 +58,8 @@ def test_mcp_client_hooks_structured_content():
 
         # Verify structured content is present and correct
         assert "structuredContent" in result
-        assert result["structuredContent"]["result"] == {"echoed": test_data}
+        assert result["structuredContent"] == {"echoed": test_data, "message_length": 15}
 
         # Verify text content matches structured content
         text_content = json.loads(result["content"][0]["text"])
-        assert text_content == {"echoed": test_data}
+        assert text_content == {"echoed": test_data, "message_length": 15}

tests_integ/mcp/test_mcp_output_schema.py

@@ -0,0 +1,44 @@
+"""Integration test for MCP tools with output schema."""
+
+from mcp import StdioServerParameters, stdio_client
+
+from strands.tools.mcp.mcp_client import MCPClient
+
+from .echo_server import EchoResponse
+
+
+def test_mcp_tool_output_schema():
+    """Test that MCP tools with output schema include it in tool spec."""
+    stdio_mcp_client = MCPClient(
+        lambda: stdio_client(StdioServerParameters(command="python", args=["tests_integ/mcp/echo_server.py"]))
+    )
+
+    with stdio_mcp_client:
+        tools = stdio_mcp_client.list_tools_sync()
+
+        # Find tools with and without output schema
+        echo_tool = next(tool for tool in tools if tool.tool_name == "echo")
+        structured_tool = next(tool for tool in tools if tool.tool_name == "echo_with_structured_content")
+
+        # Verify echo tool has no output schema
+        echo_spec = echo_tool.tool_spec
+        assert "outputSchema" not in echo_spec
+
+        # Verify structured tool has output schema
+        structured_spec = structured_tool.tool_spec
+        assert "outputSchema" in structured_spec
+
+        # Validate output schema matches expected structure
+        expected_schema = {
+            "description": "Response model for echo with structured content.",
+            "properties": {
+                "echoed": {"title": "Echoed", "type": "string"},
+                "message_length": {"title": "Message Length", "type": "integer"},
+            },
+            "required": ["echoed", "message_length"],
+            "title": "EchoResponse",
+            "type": "object",
+        }
+
+        assert structured_spec["outputSchema"]["json"] == expected_schema
+        assert structured_spec["outputSchema"]["json"] == EchoResponse.model_json_schema()