fix: address review feedback on PR strands-agents#1918

- Add echo_meta tool to echo_server.py that returns _meta from request
  context, so integration tests can verify metadata reaches the server
- Update integration tests to use echo_meta and assert the server
  received the metadata, removing the spy-based approach
- Update unit test to verify the key name is '_meta' (alias) in the
  reconstructed params dict instead of checking inject positional args

Author: Unshure

tests/strands/tools/mcp/test_mcp_instrumentation.py

@@ -452,9 +452,10 @@ def test_patch_mcp_client_preserves_existing_meta_pydantic(self):
 
             patch_function(mock_wrapped, None, [mock_request], {})
 
-            # inject should be called with the existing _meta dict (not a new empty one)
-            inject_call_args = mock_textmap_instance.inject.call_args[0][0]
-            assert inject_call_args.get("com.example/request_id") == "abc-123"
+            # Verify the reconstructed params use the key "_meta" (alias) not "meta" (Python name)
+            validated_params = mock_request.root.params.model_dump(by_alias=True)
+            assert "_meta" in validated_params
+            assert validated_params["_meta"]["com.example/request_id"] == "abc-123"
 
     def test_patch_mcp_client_injects_context_dict_params(self):
         """Test that the client patch injects OpenTelemetry context into dict params."""

tests_integ/mcp/echo_server.py

@@ -20,6 +20,7 @@
 from typing import Literal
 
 from mcp.server import FastMCP
+from mcp.server.fastmcp import Context
 from mcp.types import BlobResourceContents, CallToolResult, EmbeddedResource, TextContent, TextResourceContents
 from pydantic import BaseModel
 
@@ -48,6 +49,13 @@ def start_echo_server():
     def echo(to_echo: str) -> str:
         return to_echo
 
+    @mcp.tool(description="Echos back the _meta received in the request", structured_output=False)
+    def echo_meta(ctx: Context) -> str:
+        meta = ctx.request_context.meta
+        if meta is None:
+            return json.dumps(None)
+        return json.dumps(meta.model_dump(exclude_none=True))
+
     # 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) -> EchoResponse:

tests_integ/mcp/test_mcp_client.py

@@ -239,84 +239,62 @@ def test_mcp_client_without_structured_content():
 
 
 def test_call_tool_sync_with_meta():
-    """Test that call_tool_sync works correctly when meta is provided."""
+    """Test that call_tool_sync forwards meta to the MCP server."""
     stdio_mcp_client = MCPClient(
         lambda: stdio_client(StdioServerParameters(command="python", args=["tests_integ/mcp/echo_server.py"]))
     )
 
     with stdio_mcp_client:
         result = stdio_mcp_client.call_tool_sync(
             tool_use_id="test-meta-sync",
-            name="echo",
-            arguments={"to_echo": "META_TEST"},
+            name="echo_meta",
+            arguments={},
             meta={"com.example/request_id": "abc-123"},
         )
 
         assert result["status"] == "success"
-        assert result["content"] == [{"text": "META_TEST"}]
+        received_meta = json.loads(result["content"][0]["text"])
+        assert received_meta["com.example/request_id"] == "abc-123"
 
 
 @pytest.mark.asyncio
 async def test_call_tool_async_with_meta():
-    """Test that call_tool_async works correctly when meta is provided."""
+    """Test that call_tool_async forwards meta to the MCP server."""
     stdio_mcp_client = MCPClient(
         lambda: stdio_client(StdioServerParameters(command="python", args=["tests_integ/mcp/echo_server.py"]))
     )
 
     with stdio_mcp_client:
         result = await stdio_mcp_client.call_tool_async(
             tool_use_id="test-meta-async",
-            name="echo",
-            arguments={"to_echo": "META_ASYNC_TEST"},
+            name="echo_meta",
+            arguments={},
             meta={"com.example/request_id": "def-456"},
         )
 
         assert result["status"] == "success"
-        assert result["content"] == [{"text": "META_ASYNC_TEST"}]
+        received_meta = json.loads(result["content"][0]["text"])
+        assert received_meta["com.example/request_id"] == "def-456"
 
 
 def test_instrumentation_preserves_meta_on_tool_call():
-    """Test that OTel instrumentation correctly sets _meta on outgoing tool call requests."""
-    captured_params = []
-
-    def spy_send_request(wrapped, instance, args, kwargs):
-        if args:
-            request = args[0]
-            method = getattr(getattr(request, "root", None), "method", None)
-            if method == "tools/call" and hasattr(request.root, "params"):
-                params = request.root.params
-                if hasattr(params, "model_dump"):
-                    captured_params.append(params.model_dump(by_alias=True))
-                elif isinstance(params, dict):
-                    captured_params.append(params.copy())
-        return wrapped(*args, **kwargs)
-
+    """Test that OTel instrumentation sets _meta that reaches the MCP server."""
     stdio_mcp_client = MCPClient(
         lambda: stdio_client(StdioServerParameters(command="python", args=["tests_integ/mcp/echo_server.py"]))
     )
 
     with stdio_mcp_client:
-        from mcp.shared.session import BaseSession
-        from wrapt import wrap_function_wrapper
-
-        original_send = BaseSession.send_request
-        wrap_function_wrapper("mcp.shared.session", "BaseSession.send_request", spy_send_request)
-
-        try:
-            result = stdio_mcp_client.call_tool_sync(
-                tool_use_id="test-instrumentation",
-                name="echo",
-                arguments={"to_echo": "INSTRUMENTATION_TEST"},
-            )
-
-            assert result["status"] == "success"
-            assert len(captured_params) > 0
+        result = stdio_mcp_client.call_tool_sync(
+            tool_use_id="test-instrumentation",
+            name="echo_meta",
+            arguments={},
+        )
 
-            params = captured_params[-1]
-            assert "_meta" in params
-            assert isinstance(params["_meta"], dict)
-        finally:
-            BaseSession.send_request = original_send
+        assert result["status"] == "success"
+        received_meta = json.loads(result["content"][0]["text"])
+        # OTel instrumentation should have injected _meta with tracing context
+        assert received_meta is not None
+        assert isinstance(received_meta, dict)
 
 
 @pytest.mark.skipif(