Python: feature: Inject OpenTelemetry trace context into MCP requests and update… (#3780)

* feat: Inject OpenTelemetry trace context into MCP requests and update documentation

* Update python/samples/getting_started/observability/README.md

Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com>

* Update python/packages/core/tests/core/test_mcp.py

Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com>

* refactor: move opentelemetry import to module level

OpenTelemetry is a hard dependency of agent-framework-core (per
pyproject.toml), so the try/except ImportError guard was dead code.
Move the import to the top of the file to fail fast on missing
dependencies instead of silently hiding installation issues.

---------

Co-authored-by: Pete Roden <Pete.Roden@microsoft.com>
Co-authored-by: Mark Wallace <127216156+markwallace-microsoft@users.noreply.github.com>
Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com>

python/packages/core/agent_framework/_mcp.py

@@ -24,6 +24,7 @@ from mcp.client.websocket import websocket_client
 from mcp.shared.context import RequestContext
 from mcp.shared.exceptions import McpError
 from mcp.shared.session import RequestResponder
+from opentelemetry import propagate
 
 from ._tools import (
     FunctionTool,
@@ -380,6 +381,22 @@ def _normalize_mcp_name(name: str) -> str:
     return re.sub(r"[^A-Za-z0-9_.-]", "-", name)
 
 
+def _inject_otel_into_mcp_meta(meta: dict[str, Any] | None = None) -> dict[str, Any] | None:
+    """Inject OpenTelemetry trace context into MCP request _meta via the global propagator(s)."""
+    carrier: dict[str, str] = {}
+    propagate.inject(carrier)
+    if not carrier:
+        return meta
+
+    if meta is None:
+        meta = {}
+    for key, value in carrier.items():
+        if key not in meta:
+            meta[key] = value
+
+    return meta
+
+
 # region: MCP Plugin
 
 
@@ -875,12 +892,15 @@ class MCPTool:
             }
         }
 
+        # Inject OpenTelemetry trace context into MCP _meta for distributed tracing.
+        otel_meta = _inject_otel_into_mcp_meta()
+
         parser = self.parse_tool_results or _parse_tool_result_from_mcp
 
         # Try the operation, reconnecting once if the connection is closed
         for attempt in range(2):
             try:
-                result = await self.session.call_tool(tool_name, arguments=filtered_kwargs)  # type: ignore
+                result = await self.session.call_tool(tool_name, arguments=filtered_kwargs, meta=otel_meta)  # type: ignore
                 return parser(result)
             except ClosedResourceError as cl_ex:
                 if attempt == 0:

python/packages/core/tests/core/test_mcp.py

@@ -2591,3 +2591,70 @@ async def test_mcp_tool_filters_framework_kwargs():
         assert "thread" not in arguments
         assert "conversation_id" not in arguments
         assert "options" not in arguments
+
+
+# region: OTel trace context propagation via _meta
+
+
+@pytest.mark.parametrize(
+    "use_span,expect_traceparent",
+    [
+        (True, True),
+        (False, False),
+    ],
+)
+async def test_mcp_tool_call_tool_otel_meta(use_span, expect_traceparent, span_exporter):
+    """call_tool propagates OTel trace context via meta only when a span is active."""
+    from opentelemetry import trace
+
+    class TestServer(MCPTool):
+        async def connect(self):
+            self.session = Mock(spec=ClientSession)
+            self.session.list_tools = AsyncMock(
+                return_value=types.ListToolsResult(
+                    tools=[
+                        types.Tool(
+                            name="test_tool",
+                            description="Test tool",
+                            inputSchema={
+                                "type": "object",
+                                "properties": {"param": {"type": "string"}},
+                                "required": ["param"],
+                            },
+                        )
+                    ]
+                )
+            )
+            self.session.call_tool = AsyncMock(
+                return_value=types.CallToolResult(content=[types.TextContent(type="text", text="result")])
+            )
+
+        def get_mcp_client(self) -> _AsyncGeneratorContextManager[Any, None]:
+            return None
+
+    server = TestServer(name="test_server")
+    async with server:
+        await server.load_tools()
+
+        if use_span:
+            tracer = trace.get_tracer("test")
+            with tracer.start_as_current_span("test_span"):
+                await server.functions[0].invoke(param="test_value")
+        else:
+            # Use an invalid span to ensure no trace context is injected;
+            # call server.call_tool directly to bypass FunctionTool.invoke's own span.
+            with trace.use_span(trace.NonRecordingSpan(trace.INVALID_SPAN_CONTEXT)):
+                await server.call_tool("test_tool", param="test_value")
+
+        meta = server.session.call_tool.call_args.kwargs.get("meta")
+        if expect_traceparent:
+            # When a valid span is active, we expect some propagation fields to be injected,
+            # but we do not assume any specific header name to keep this test propagator-agnostic.
+            assert meta is not None
+            assert isinstance(meta, dict)
+            assert len(meta) > 0
+        else:
+            assert meta is None
+
+
+# endregion