Add basic OpenTelemetry tracing for client and server requests (#2381)

Author: Kludex

pyproject.toml

@@ -40,6 +40,7 @@ dependencies = [
     "pyjwt[crypto]>=2.10.1",
     "typing-extensions>=4.13.0",
     "typing-inspection>=0.4.1",
+    "opentelemetry-api>=1.28.0",
 ]
 
 [project.optional-dependencies]
@@ -71,6 +72,7 @@ dev = [
     "coverage[toml]>=7.10.7,<=7.13",
     "pillow>=12.0",
     "strict-no-cover",
+    "logfire>=3.0.0",
 ]
 docs = [
     "mkdocs>=1.6.1",

src/mcp/server/lowlevel/server.py

@@ -42,9 +42,10 @@ async def main():
 from collections.abc import AsyncIterator, Awaitable, Callable
 from contextlib import AbstractAsyncContextManager, AsyncExitStack, asynccontextmanager
 from importlib.metadata import version as importlib_version
-from typing import Any, Generic
+from typing import Any, Generic, cast
 
 import anyio
+from opentelemetry.trace import SpanKind, StatusCode
 from starlette.applications import Starlette
 from starlette.middleware import Middleware
 from starlette.middleware.authentication import AuthenticationMiddleware
@@ -65,6 +66,7 @@ async def main():
 from mcp.server.streamable_http import EventStore
 from mcp.server.streamable_http_manager import StreamableHTTPASGIApp, StreamableHTTPSessionManager
 from mcp.server.transport_security import TransportSecuritySettings
+from mcp.shared._otel import extract_trace_context, otel_span
 from mcp.shared._stream_protocols import ReadStream, WriteStream
 from mcp.shared.exceptions import MCPError
 from mcp.shared.message import ServerMessageMetadata, SessionMessage
@@ -446,72 +448,90 @@ async def _handle_request(
     ):
         logger.info("Processing request of type %s", type(req).__name__)
 
-        if handler := self._request_handlers.get(req.method):
-            logger.debug("Dispatching request of type %s", type(req).__name__)
+        target = getattr(req.params, "name", None) if req.params else None
+        span_name = f"MCP handle {req.method} {target}" if target else f"MCP handle {req.method}"
 
-            try:
-                # Extract request context and close_sse_stream from message metadata
-                request_data = None
-                close_sse_stream_cb = None
-                close_standalone_sse_stream_cb = None
-                if message.message_metadata is not None and isinstance(message.message_metadata, ServerMessageMetadata):
-                    request_data = message.message_metadata.request_context
-                    close_sse_stream_cb = message.message_metadata.close_sse_stream
-                    close_standalone_sse_stream_cb = message.message_metadata.close_standalone_sse_stream
+        # Extract W3C trace context from _meta (SEP-414).
+        meta = cast(dict[str, Any] | None, getattr(req.params, "meta", None)) if req.params else None
+        parent_context = extract_trace_context(meta) if meta is not None else None
 
-                client_capabilities = session.client_params.capabilities if session.client_params else None
-                task_support = self._experimental_handlers.task_support if self._experimental_handlers else None
-                # Get task metadata from request params if present
-                task_metadata = None
-                if hasattr(req, "params") and req.params is not None:
-                    task_metadata = getattr(req.params, "task", None)
-                ctx = ServerRequestContext(
-                    request_id=message.request_id,
-                    meta=message.request_meta,
-                    session=session,
-                    lifespan_context=lifespan_context,
-                    experimental=Experimental(
-                        task_metadata=task_metadata,
-                        _client_capabilities=client_capabilities,
-                        _session=session,
-                        _task_support=task_support,
-                    ),
-                    request=request_data,
-                    close_sse_stream=close_sse_stream_cb,
-                    close_standalone_sse_stream=close_standalone_sse_stream_cb,
-                )
-                response = await handler(ctx, req.params)
-            except MCPError as err:
-                response = err.error
-            except anyio.get_cancelled_exc_class():
-                if message.cancelled:
-                    # Client sent CancelledNotification; responder.cancel() already
-                    # sent an error response, so skip the duplicate.
-                    logger.info("Request %s cancelled - duplicate response suppressed", message.request_id)
-                    return
-                # Transport-close cancellation from the TG in run(); re-raise so the
-                # TG swallows its own cancellation.
-                raise
-            except Exception as err:
-                if raise_exceptions:  # pragma: no cover
-                    raise err
-                response = types.ErrorData(code=0, message=str(err))
-        else:  # pragma: no cover
-            response = types.ErrorData(code=types.METHOD_NOT_FOUND, message="Method not found")
-
-        try:
-            await message.respond(response)
-        except (anyio.BrokenResourceError, anyio.ClosedResourceError):
-            # Transport closed between handler unblocking and respond. Happens
-            # when _receive_loop's finally wakes a handler blocked on
-            # send_request: the handler runs to respond() before run()'s TG
-            # cancel fires, but after the write stream closed. Closed if our
-            # end closed (_receive_loop's async-with exit); Broken if the peer
-            # end closed first (streamable_http terminate()).
-            logger.debug("Response for %s dropped - transport closed", message.request_id)
-            return
-
-        logger.debug("Response sent")
+        with otel_span(
+            span_name,
+            kind=SpanKind.SERVER,
+            attributes={"mcp.method.name": req.method, "jsonrpc.request.id": message.request_id},
+            context=parent_context,
+        ) as span:
+            if handler := self._request_handlers.get(req.method):
+                logger.debug("Dispatching request of type %s", type(req).__name__)
+
+                try:
+                    # Extract request context and close_sse_stream from message metadata
+                    request_data = None
+                    close_sse_stream_cb = None
+                    close_standalone_sse_stream_cb = None
+                    if message.message_metadata is not None and isinstance(
+                        message.message_metadata, ServerMessageMetadata
+                    ):
+                        request_data = message.message_metadata.request_context
+                        close_sse_stream_cb = message.message_metadata.close_sse_stream
+                        close_standalone_sse_stream_cb = message.message_metadata.close_standalone_sse_stream
+
+                    client_capabilities = session.client_params.capabilities if session.client_params else None
+                    task_support = self._experimental_handlers.task_support if self._experimental_handlers else None
+                    # Get task metadata from request params if present
+                    task_metadata = None
+                    if hasattr(req, "params") and req.params is not None:  # pragma: no branch
+                        task_metadata = getattr(req.params, "task", None)
+                    ctx = ServerRequestContext(
+                        request_id=message.request_id,
+                        meta=message.request_meta,
+                        session=session,
+                        lifespan_context=lifespan_context,
+                        experimental=Experimental(
+                            task_metadata=task_metadata,
+                            _client_capabilities=client_capabilities,
+                            _session=session,
+                            _task_support=task_support,
+                        ),
+                        request=request_data,
+                        close_sse_stream=close_sse_stream_cb,
+                        close_standalone_sse_stream=close_standalone_sse_stream_cb,
+                    )
+                    response = await handler(ctx, req.params)
+                except MCPError as err:
+                    response = err.error
+                except anyio.get_cancelled_exc_class():
+                    if message.cancelled:
+                        # Client sent CancelledNotification; responder.cancel() already
+                        # sent an error response, so skip the duplicate.
+                        logger.info("Request %s cancelled - duplicate response suppressed", message.request_id)
+                        return
+                    # Transport-close cancellation from the TG in run(); re-raise so the
+                    # TG swallows its own cancellation.
+                    raise
+                except Exception as err:
+                    if raise_exceptions:  # pragma: no cover
+                        raise err
+                    response = types.ErrorData(code=0, message=str(err))
+            else:  # pragma: no cover
+                response = types.ErrorData(code=types.METHOD_NOT_FOUND, message="Method not found")
+
+            if isinstance(response, types.ErrorData) and span is not None:
+                span.set_status(StatusCode.ERROR, response.message)
+
+            try:
+                await message.respond(response)
+            except (anyio.BrokenResourceError, anyio.ClosedResourceError):
+                # Transport closed between handler unblocking and respond. Happens
+                # when _receive_loop's finally wakes a handler blocked on
+                # send_request: the handler runs to respond() before run()'s TG
+                # cancel fires, but after the write stream closed. Closed if our
+                # end closed (_receive_loop's async-with exit); Broken if the peer
+                # end closed first (streamable_http terminate()).
+                logger.debug("Response for %s dropped - transport closed", message.request_id)
+                return
+
+            logger.debug("Response sent")
 
     async def _handle_notification(
         self,

src/mcp/shared/_otel.py

@@ -0,0 +1,36 @@
+"""OpenTelemetry helpers for MCP."""
+
+from __future__ import annotations
+
+from collections.abc import Iterator
+from contextlib import contextmanager
+from typing import Any
+
+from opentelemetry.context import Context
+from opentelemetry.propagate import extract, inject
+from opentelemetry.trace import SpanKind, get_tracer
+
+_tracer = get_tracer("mcp-python-sdk")
+
+
+@contextmanager
+def otel_span(
+    name: str,
+    *,
+    kind: SpanKind,
+    attributes: dict[str, Any] | None = None,
+    context: Context | None = None,
+) -> Iterator[Any]:
+    """Create an OTel span."""
+    with _tracer.start_as_current_span(name, kind=kind, attributes=attributes, context=context) as span:
+        yield span
+
+
+def inject_trace_context(meta: dict[str, Any]) -> None:
+    """Inject W3C trace context (traceparent/tracestate) into a `_meta` dict."""
+    inject(meta)
+
+
+def extract_trace_context(meta: dict[str, Any]) -> Context:
+    """Extract W3C trace context from a `_meta` dict."""
+    return extract(meta)

src/mcp/shared/session.py

@@ -9,9 +9,11 @@
 
 import anyio
 from anyio.streams.memory import MemoryObjectSendStream
+from opentelemetry.trace import SpanKind
 from pydantic import BaseModel, TypeAdapter
 from typing_extensions import Self
 
+from mcp.shared._otel import inject_trace_context, otel_span
 from mcp.shared._stream_protocols import ReadStream, WriteStream
 from mcp.shared.exceptions import MCPError
 from mcp.shared.message import MessageMetadata, ServerMessageMetadata, SessionMessage
@@ -268,24 +270,36 @@ async def send_request(
             self._progress_callbacks[request_id] = progress_callback
 
         try:
-            jsonrpc_request = JSONRPCRequest(jsonrpc="2.0", id=request_id, **request_data)
-            await self._write_stream.send(SessionMessage(message=jsonrpc_request, metadata=metadata))
-
-            # request read timeout takes precedence over session read timeout
-            timeout = request_read_timeout_seconds or self._session_read_timeout_seconds
-
-            try:
-                with anyio.fail_after(timeout):
-                    response_or_error = await response_stream_reader.receive()
-            except TimeoutError:
-                class_name = request.__class__.__name__
-                message = f"Timed out while waiting for response to {class_name}. Waited {timeout} seconds."
-                raise MCPError(code=REQUEST_TIMEOUT, message=message)
-
-            if isinstance(response_or_error, JSONRPCError):
-                raise MCPError.from_jsonrpc_error(response_or_error)
-            else:
-                return result_type.model_validate(response_or_error.result, by_name=False)
+            target = request_data.get("params", {}).get("name")
+            span_name = f"MCP send {request.method} {target}" if target else f"MCP send {request.method}"
+
+            with otel_span(
+                span_name,
+                kind=SpanKind.CLIENT,
+                attributes={"mcp.method.name": request.method, "jsonrpc.request.id": request_id},
+            ):
+                # Inject W3C trace context into _meta (SEP-414).
+                meta: dict[str, Any] = request_data.setdefault("params", {}).setdefault("_meta", {})
+                inject_trace_context(meta)
+
+                jsonrpc_request = JSONRPCRequest(jsonrpc="2.0", id=request_id, **request_data)
+                await self._write_stream.send(SessionMessage(message=jsonrpc_request, metadata=metadata))
+
+                # request read timeout takes precedence over session read timeout
+                timeout = request_read_timeout_seconds or self._session_read_timeout_seconds
+
+                try:
+                    with anyio.fail_after(timeout):
+                        response_or_error = await response_stream_reader.receive()
+                except TimeoutError:
+                    class_name = request.__class__.__name__
+                    message = f"Timed out while waiting for response to {class_name}. Waited {timeout} seconds."
+                    raise MCPError(code=REQUEST_TIMEOUT, message=message)
+
+                if isinstance(response_or_error, JSONRPCError):
+                    raise MCPError.from_jsonrpc_error(response_or_error)
+                else:
+                    return result_type.model_validate(response_or_error.result, by_name=False)
 
         finally:
             self._response_streams.pop(request_id, None)

tests/shared/test_otel.py

@@ -0,0 +1,44 @@
+from __future__ import annotations
+
+import pytest
+from logfire.testing import CaptureLogfire
+
+from mcp import types
+from mcp.client.client import Client
+from mcp.server.mcpserver import MCPServer
+
+pytestmark = pytest.mark.anyio
+
+
+# Logfire warns about propagated trace context by default (distributed_tracing=None).
+# This is expected here since we're testing cross-boundary context propagation.
+@pytest.mark.filterwarnings("ignore::RuntimeWarning")
+async def test_client_and_server_spans(capfire: CaptureLogfire):
+    """Verify that calling a tool produces client and server spans with correct attributes."""
+    server = MCPServer("test")
+
+    @server.tool()
+    def greet(name: str) -> str:
+        """Greet someone."""
+        return f"Hello, {name}!"
+
+    async with Client(server) as client:
+        result = await client.call_tool("greet", {"name": "World"})
+
+    assert isinstance(result.content[0], types.TextContent)
+    assert result.content[0].text == "Hello, World!"
+
+    spans = capfire.exporter.exported_spans_as_dict()
+    span_names = {s["name"] for s in spans}
+
+    assert "MCP send tools/call greet" in span_names
+    assert "MCP handle tools/call greet" in span_names
+
+    client_span = next(s for s in spans if s["name"] == "MCP send tools/call greet")
+    server_span = next(s for s in spans if s["name"] == "MCP handle tools/call greet")
+
+    assert client_span["attributes"]["mcp.method.name"] == "tools/call"
+    assert server_span["attributes"]["mcp.method.name"] == "tools/call"
+
+    # Server span should be in the same trace as the client span (context propagation).
+    assert server_span["context"]["trace_id"] == client_span["context"]["trace_id"]