feat: add meta field support to MCP tool results

src/strands/tools/mcp/mcp_client.py

@@ -566,6 +566,8 @@ def _handle_tool_result(self, tool_use_id: str, call_tool_result: MCPCallToolRes
 
         if call_tool_result.structuredContent:
             result["structuredContent"] = call_tool_result.structuredContent
+        if call_tool_result.meta:
+            result["metadata"] = call_tool_result.meta
 
         return result
 

src/strands/tools/mcp/mcp_types.py

@@ -1,7 +1,7 @@
 """Type definitions for MCP integration."""
 
 from contextlib import AbstractAsyncContextManager
-from typing import Any, Dict
+from typing import Any
 
 from anyio.streams.memory import MemoryObjectReceiveStream, MemoryObjectSendStream
 from mcp.client.streamable_http import GetSessionIdCallback
@@ -58,6 +58,10 @@ class MCPToolResult(ToolResult):
         structuredContent: Optional JSON object containing structured data returned
             by the MCP tool. This allows MCP tools to return complex data structures
             that can be processed programmatically by agents or other tools.
+        metadata: Optional arbitrary metadata returned by the MCP tool. This field allows
+            MCP servers to attach custom metadata to tool results (e.g., token usage,
+            performance metrics, or business-specific tracking information).
     """
 
-    structuredContent: NotRequired[Dict[str, Any]]
+    structuredContent: NotRequired[dict[str, Any]]
+    metadata: NotRequired[dict[str, Any]]

tests/strands/tools/mcp/test_mcp_client.py

@@ -533,7 +533,7 @@ def test_stop_closes_event_loop():
     mock_thread.join = MagicMock()
     mock_event_loop = MagicMock()
     mock_event_loop.close = MagicMock()
-    
+
     client._background_thread = mock_thread
     client._background_thread_event_loop = mock_event_loop
 
@@ -542,7 +542,7 @@ def test_stop_closes_event_loop():
 
     # Verify thread was joined
     mock_thread.join.assert_called_once()
-    
+
     # Verify event loop was closed
     mock_event_loop.close.assert_called_once()
 
@@ -750,3 +750,65 @@ async def test_handle_error_message_non_exception():
 
     # This should not raise an exception
     await client._handle_error_message("normal message")
+
+
+def test_call_tool_sync_with_meta_field(mock_transport, mock_session):
+    """Test that call_tool_sync correctly handles meta field."""
+    mock_content = MCPTextContent(type="text", text="Test message")
+    meta_data = {"tokenUsage": {"inputTokens": 100, "outputTokens": 50}, "executionTime": 1.5}
+    mock_session.call_tool.return_value = MCPCallToolResult(isError=False, content=[mock_content], meta=meta_data)
+
+    with MCPClient(mock_transport["transport_callable"]) as client:
+        result = client.call_tool_sync(tool_use_id="test-123", name="test_tool", arguments={"param": "value"})
+
+        mock_session.call_tool.assert_called_once_with("test_tool", {"param": "value"}, None)
+
+        assert result["status"] == "success"
+        assert result["toolUseId"] == "test-123"
+        assert len(result["content"]) == 1
+        assert result["content"][0]["text"] == "Test message"
+        assert "meta" in result
+        assert result["meta"] == meta_data
+        assert result["meta"]["tokenUsage"]["inputTokens"] == 100
+        assert result["meta"]["tokenUsage"]["outputTokens"] == 50
+        assert result["meta"]["executionTime"] == 1.5
+
+
+def test_call_tool_sync_without_meta_field(mock_transport, mock_session):
+    """Test that call_tool_sync works correctly when no meta field is provided."""
+    mock_content = MCPTextContent(type="text", text="Test message")
+    mock_session.call_tool.return_value = MCPCallToolResult(
+        isError=False,
+        content=[mock_content],
+    )
+
+    with MCPClient(mock_transport["transport_callable"]) as client:
+        result = client.call_tool_sync(tool_use_id="test-123", name="test_tool", arguments={"param": "value"})
+
+        assert result["status"] == "success"
+        assert result["toolUseId"] == "test-123"
+        assert len(result["content"]) == 1
+        assert result["content"][0]["text"] == "Test message"
+        assert result.get("meta") is None
+
+
+def test_call_tool_sync_with_meta_and_structured_content(mock_transport, mock_session):
+    """Test that call_tool_sync correctly handles both meta and structuredContent fields."""
+    mock_content = MCPTextContent(type="text", text="Test message")
+    meta_data = {"tokenUsage": {"inputTokens": 100, "outputTokens": 50}}
+    structured_content = {"result": 42, "status": "completed"}
+    mock_session.call_tool.return_value = MCPCallToolResult(
+        isError=False, content=[mock_content], meta=meta_data, structuredContent=structured_content
+    )
+
+    with MCPClient(mock_transport["transport_callable"]) as client:
+        result = client.call_tool_sync(tool_use_id="test-123", name="test_tool", arguments={"param": "value"})
+
+        mock_session.call_tool.assert_called_once_with("test_tool", {"param": "value"}, None)
+
+        assert result["status"] == "success"
+        assert result["toolUseId"] == "test-123"
+        assert "meta" in result
+        assert result["meta"] == meta_data
+        assert "structuredContent" in result
+        assert result["structuredContent"] == structured_content

tests_integ/mcp/echo_server.py

@@ -19,7 +19,7 @@
 from typing import Literal
 
 from mcp.server import FastMCP
-from mcp.types import BlobResourceContents, EmbeddedResource, TextResourceContents
+from mcp.types import BlobResourceContents, CallToolResult, EmbeddedResource, TextContent, TextResourceContents
 from pydantic import BaseModel
 
 
@@ -50,6 +50,16 @@ def echo(to_echo: str) -> str:
     def echo_with_structured_content(to_echo: str) -> EchoResponse:
         return EchoResponse(echoed=to_echo, message_length=len(to_echo))
 
+    @mcp.tool(description="Echos response back with metadata")
+    def echo_with_metadata(to_echo: str):
+        """Return structured content and metadata in the tool result."""
+
+        return CallToolResult(
+            content=[TextContent(type="text", text=to_echo)],
+            isError=False,
+            _meta={"metadata": {"nested": 1}, "shallow": "val"},
+        )
+
     @mcp.tool(description="Get current weather information for a location")
     def get_weather(location: Literal["New York", "London", "Tokyo"] = "New York"):
         """Get weather data including forecasts and alerts for the specified location"""

tests_integ/mcp/test_mcp_client_structured_content_and_metadata.py

@@ -0,0 +1,95 @@
+"""Integration test for MCP client structured content and metadata support.
+
+This test verifies that MCP tools can return structured content and metadata,
+and that the MCP client properly handles and exposes these fields in tool results.
+"""
+
+import json
+
+from mcp import StdioServerParameters, stdio_client
+
+from strands import Agent
+from strands.hooks import AfterToolCallEvent, HookProvider, HookRegistry
+from strands.tools.mcp.mcp_client import MCPClient
+
+
+class ToolResultCapture(HookProvider):
+    """Captures tool results for inspection."""
+
+    def __init__(self):
+        self.captured_results = {}
+
+    def register_hooks(self, registry: HookRegistry) -> None:
+        """Register callback for after tool invocation events."""
+        registry.add_callback(AfterToolCallEvent, self.on_after_tool_invocation)
+
+    def on_after_tool_invocation(self, event: AfterToolCallEvent) -> None:
+        """Capture tool results by tool name."""
+        tool_name = event.tool_use["name"]
+        self.captured_results[tool_name] = event.result
+
+
+def test_structured_content():
+    """Test that MCP tools can return structured content."""
+    # Set up result capture
+    result_capture = ToolResultCapture()
+
+    # Set up MCP client for echo server
+    stdio_mcp_client = MCPClient(
+        lambda: stdio_client(StdioServerParameters(command="python", args=["tests_integ/mcp/echo_server.py"]))
+    )
+
+    with stdio_mcp_client:
+        # Create agent with MCP tools and result capture
+        agent = Agent(tools=stdio_mcp_client.list_tools_sync(), hooks=[result_capture])
+
+        # Test structured content functionality
+        test_data = "STRUCTURED_TEST"
+        agent(f"Use the echo_with_structured_content tool to echo: {test_data}")
+
+        # Verify result was captured
+        assert "echo_with_structured_content" in result_capture.captured_results
+        result = result_capture.captured_results["echo_with_structured_content"]
+
+        # Verify basic result structure
+        assert result["status"] == "success"
+        assert len(result["content"]) == 1
+
+        # Verify structured content is present and correct
+        assert "structuredContent" in result
+        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, "message_length": 15}
+
+
+def test_metadata():
+    """Test that MCP tools can return metadata."""
+    # Set up result capture
+    result_capture = ToolResultCapture()
+
+    # Set up MCP client for echo server
+    stdio_mcp_client = MCPClient(
+        lambda: stdio_client(StdioServerParameters(command="python", args=["tests_integ/mcp/echo_server.py"]))
+    )
+
+    with stdio_mcp_client:
+        # Create agent with MCP tools and result capture
+        agent = Agent(tools=stdio_mcp_client.list_tools_sync(), hooks=[result_capture])
+
+        # Test metadata functionality
+        test_data = "METADATA_TEST"
+        agent(f"Use the echo_with_metadata tool to echo: {test_data}")
+
+        # Verify result was captured
+        assert "echo_with_metadata" in result_capture.captured_results
+        result = result_capture.captured_results["echo_with_metadata"]
+
+        # Verify basic result structure
+        assert result["status"] == "success"
+
+        # Verify metadata is present and correct
+        assert "metadata" in result
+        expected_metadata = {"metadata": {"nested": 1}, "shallow": "val"}
+        assert result["metadata"] == expected_metadata