Skip to content
Closed
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
3 changes: 3 additions & 0 deletions src/strands/tools/mcp/mcp_client.py
Original file line number Diff line number Diff line change
Expand Up @@ -567,6 +567,9 @@ 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["meta"] = call_tool_result.meta

return result

async def _async_background_thread(self) -> None:
Expand Down
4 changes: 4 additions & 0 deletions src/strands/tools/mcp/mcp_types.py
Original file line number Diff line number Diff line change
Expand Up @@ -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.
meta: Optional JSON object containing metadata about the tool execution
returned by the MCP tool. This provides additional context or information
about how the tool was executed or processed.
"""

structuredContent: NotRequired[Dict[str, Any]]
meta: NotRequired[Dict[str, Any]]
97 changes: 95 additions & 2 deletions tests/strands/tools/mcp/test_mcp_client.py
Original file line number Diff line number Diff line change
Expand Up @@ -358,13 +358,35 @@ def test_mcp_tool_result_type():
# Test that structuredContent is optional
assert "structuredContent" not in result or result.get("structuredContent") is None

# Test that meta is optional
assert "meta" not in result or result.get("meta") is None

# Test with structuredContent
result_with_structured = MCPToolResult(
status="success", toolUseId="test-456", content=[{"text": "Test message"}], structuredContent={"key": "value"}
)

assert result_with_structured["structuredContent"] == {"key": "value"}

# Test with meta
result_with_meta = MCPToolResult(
status="success", toolUseId="test-789", content=[{"text": "Test message"}], meta={"request_id": "req123"}
)

assert result_with_meta["meta"] == {"request_id": "req123"}

# Test with both structuredContent and meta
result_with_both = MCPToolResult(
status="success",
toolUseId="test-999",
content=[{"text": "Test message"}],
structuredContent={"result": "data"},
meta={"request_id": "req456"},
)

assert result_with_both["structuredContent"] == {"result": "data"}
assert result_with_both["meta"] == {"request_id": "req456"}


def test_call_tool_sync_without_structured_content(mock_transport, mock_session):
"""Test that call_tool_sync works correctly when no structured content is provided."""
Expand All @@ -385,6 +407,77 @@ def test_call_tool_sync_without_structured_content(mock_transport, mock_session)
assert result.get("structuredContent") is None


def test_call_tool_sync_with_meta(mock_transport, mock_session):
"""Test that call_tool_sync correctly handles meta field."""
mock_content = MCPTextContent(type="text", text="Test message")
meta_data = {"request_id": "abc123", "timestamp": 1234567890}
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"
# Content should only contain the text content, not the meta
assert len(result["content"]) == 1
assert result["content"][0]["text"] == "Test message"
# Meta should be in its own field
assert "meta" in result
assert result["meta"] == meta_data
assert result["meta"]["request_id"] == "abc123"
assert result["meta"]["timestamp"] == 1234567890


def test_call_tool_sync_without_meta(mock_transport, mock_session):
"""Test that call_tool_sync works correctly when no meta is provided."""
mock_content = MCPTextContent(type="text", text="Test message")
mock_session.call_tool.return_value = MCPCallToolResult(
isError=False,
content=[mock_content], # No meta
)

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"
# meta should be None when not provided by MCP
assert result.get("meta") is None


def test_call_tool_sync_with_structured_content_and_meta(mock_transport, mock_session):
"""Test that call_tool_sync correctly handles both structured content and meta."""
mock_content = MCPTextContent(type="text", text="Test message")
structured_content = {"result": 42, "status": "completed"}
meta_data = {"request_id": "xyz789", "processing_time_ms": 150}
mock_session.call_tool.return_value = MCPCallToolResult(
isError=False, content=[mock_content], structuredContent=structured_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"
# Content should only contain the text content
assert len(result["content"]) == 1
assert result["content"][0]["text"] == "Test message"
# Structured content should be in its own field
assert "structuredContent" in result
assert result["structuredContent"] == structured_content
# Meta should be in its own field
assert "meta" in result
assert result["meta"] == meta_data
assert result["meta"]["request_id"] == "xyz789"
assert result["meta"]["processing_time_ms"] == 150


def test_exception_when_future_not_running():
"""Test exception handling when the future is not running."""
# Create a client.with a mock transport
Expand Down Expand Up @@ -533,7 +626,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

Expand All @@ -542,7 +635,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()

Expand Down
22 changes: 21 additions & 1 deletion tests_integ/mcp/echo_server.py
Original file line number Diff line number Diff line change
Expand Up @@ -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


Expand Down Expand Up @@ -50,6 +50,26 @@ 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 meta field")
def echo_with_meta(to_echo: str) -> CallToolResult:
"""Echo tool that returns CallToolResult with meta field."""
return CallToolResult(
content=[TextContent(type="text", text=to_echo)],
isError=False,
_meta={"request_id": "test-request-123", "echo_length": len(to_echo)},
)

@mcp.tool(description="Echos response back with both structured content and meta", structured_output=True)
def echo_with_structured_content_and_meta(to_echo: str) -> CallToolResult:
"""Echo tool that returns CallToolResult with both structured content and meta."""
response = EchoResponse(echoed=to_echo, message_length=len(to_echo))
return CallToolResult(
content=[TextContent(type="text", text=response.model_dump_json())],
structuredContent=response.model_dump(),
isError=False,
_meta={"request_id": "test-request-456", "processing_time_ms": 100},
)

@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"""
Expand Down
132 changes: 132 additions & 0 deletions tests_integ/mcp/test_mcp_client_meta_with_hooks.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,132 @@
"""Integration test demonstrating MCP client meta field support.

This test shows how the MCP client properly handles the meta field returned by
MCP tools, both with and without structured content.
"""

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 MetaHookProvider(HookProvider):
"""Hook provider that captures tool results with meta field."""

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."""
tool_name = event.tool_use["name"]
self.captured_results[tool_name] = event.result


def test_mcp_client_with_meta_only():
"""Test that MCP client correctly handles tools that return meta without structured content."""
# Create hook provider to capture tool result
hook_provider = MetaHookProvider()

# 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 hook provider
agent = Agent(tools=stdio_mcp_client.list_tools_sync(), hooks=[hook_provider])

# Test meta field functionality
test_data = "META_TEST_DATA"
agent(f"Use the echo_with_meta tool to echo: {test_data}")

# Verify hook captured the tool result
assert "echo_with_meta" in hook_provider.captured_results
result = hook_provider.captured_results["echo_with_meta"]

# Verify basic result structure
assert result["status"] == "success"
assert len(result["content"]) == 1
assert result["content"][0]["text"] == test_data

# Verify meta is present and correct
assert "meta" in result
assert result["meta"]["request_id"] == "test-request-123"
assert result["meta"]["echo_length"] == len(test_data)

# Verify structured content is not present
assert result.get("structuredContent") is None


def test_mcp_client_with_structured_content_and_meta():
"""Test that MCP client correctly handles tools that return both structured content and meta."""
# Create hook provider to capture tool result
hook_provider = MetaHookProvider()

# 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 hook provider
agent = Agent(tools=stdio_mcp_client.list_tools_sync(), hooks=[hook_provider])

# Test structured content and meta functionality
test_data = "BOTH_TEST_DATA"
agent(f"Use the echo_with_structured_content_and_meta tool to echo: {test_data}")

# Verify hook captured the tool result
assert "echo_with_structured_content_and_meta" in hook_provider.captured_results
result = hook_provider.captured_results["echo_with_structured_content_and_meta"]

# Verify basic result structure
assert result["status"] == "success"
assert len(result["content"]) == 1

# Verify structured content is present
assert "structuredContent" in result
assert result["structuredContent"]["echoed"] == test_data
assert result["structuredContent"]["message_length"] == len(test_data)

# Verify meta is present and correct
assert "meta" in result
assert result["meta"]["request_id"] == "test-request-456"
assert result["meta"]["processing_time_ms"] == 100


def test_mcp_client_without_meta():
"""Test that MCP client works correctly when tool does not return meta."""
# Create hook provider to capture tool result
hook_provider = MetaHookProvider()

# 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 hook provider
agent = Agent(tools=stdio_mcp_client.list_tools_sync(), hooks=[hook_provider])

# Test regular echo tool (no meta, no structured content)
test_data = "SIMPLE_TEST_DATA"
agent(f"Use the echo tool to echo: {test_data}")

# Verify hook captured the tool result
assert "echo" in hook_provider.captured_results
result = hook_provider.captured_results["echo"]

# Verify basic result structure
assert result["status"] == "success"
assert len(result["content"]) == 1
assert result["content"][0]["text"] == test_data

# Verify neither meta nor structured content is present
assert result.get("meta") is None
assert result.get("structuredContent") is None