Add basic OpenTelemetry tracing for client and server requests

Add opentelemetry-api as an optional dependency (mcp[otel]) and create
spans for client request/response cycles and server request handling.

Span names follow the pattern:
- Client: "MCP {method} {target}" (e.g. "MCP tools/call my_tool")
- Server: "MCP handle {method} {target}"

When opentelemetry-api is not installed, tracing is a complete no-op
with zero overhead.

Author: Kludex

pyproject.toml

@@ -46,6 +46,7 @@ dependencies = [
 rich = ["rich>=13.9.4"]
 cli = ["typer>=0.16.0", "python-dotenv>=1.0.0"]
 ws = ["websockets>=15.0.1"]
+otel = ["opentelemetry-api>=1.28.0"]
 
 [project.scripts]
 mcp = "mcp.cli:app [cli]"
@@ -57,7 +58,7 @@ required-version = ">=0.9.5"
 [dependency-groups]
 dev = [
     # We add mcp[cli,ws] so `uv sync` considers the extras.
-    "mcp[cli,ws]",
+    "mcp[cli,ws,otel]",
     "pyright>=1.1.400",
     "pytest>=8.3.4",
     "ruff>=0.8.5",

src/mcp/server/lowlevel/server.py

@@ -65,6 +65,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 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 +447,82 @@ 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
+        with otel_span(
+            span_name,
+            kind="SERVER",
+            attributes={"mcp.method.name": req.method, "jsonrpc.request.id": message.request_id},
+        ):
+            if handler := self._request_handlers.get(req.method):
+                logger.debug("Dispatching request of type %s", type(req).__name__)
 
-                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")
+                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:
+                        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")
 
     async def _handle_notification(
         self,

src/mcp/shared/_otel.py

@@ -0,0 +1,43 @@
+"""OpenTelemetry helpers for MCP.
+
+Provides a context manager that creates an OpenTelemetry span when
+``opentelemetry-api`` is installed, or acts as a no-op otherwise.
+"""
+
+from __future__ import annotations
+
+import functools
+from collections.abc import Iterator
+from contextlib import contextmanager
+from typing import Any
+
+
+@functools.lru_cache(maxsize=1)
+def _get_tracer() -> Any:
+    """Return the OTel tracer for ``mcp``, or ``None``."""
+    try:
+        from opentelemetry.trace import get_tracer
+
+        return get_tracer("mcp-python-sdk")
+    except ImportError:
+        return None
+
+
+@contextmanager
+def otel_span(
+    name: str,
+    *,
+    kind: str = "INTERNAL",
+    attributes: dict[str, Any] | None = None,
+) -> Iterator[Any]:
+    """Create an OTel span if ``opentelemetry-api`` is installed, else no-op."""
+    tracer = _get_tracer()
+    if tracer is None:
+        yield None
+        return
+
+    from opentelemetry.trace import SpanKind
+
+    span_kind = getattr(SpanKind, kind, SpanKind.INTERNAL)
+    with tracer.start_as_current_span(name, kind=span_kind, attributes=attributes) as span:
+        yield span

src/mcp/shared/session.py

@@ -12,6 +12,7 @@
 from pydantic import BaseModel, TypeAdapter
 from typing_extensions import Self
 
+from mcp.shared._otel import otel_span
 from mcp.shared._stream_protocols import ReadStream, WriteStream
 from mcp.shared.exceptions import MCPError
 from mcp.shared.message import MessageMetadata, ServerMessageMetadata, SessionMessage
@@ -269,23 +270,32 @@ async def send_request(
 
         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 {request.method} {target}" if target else f"MCP {request.method}"
+
+            with otel_span(
+                span_name,
+                kind="CLIENT",
+                attributes={"mcp.method.name": request.method, "jsonrpc.request.id": request_id},
+            ):
+                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,24 @@
+from __future__ import annotations
+
+from unittest.mock import patch
+
+import pytest
+
+from mcp.shared._otel import _get_tracer, otel_span
+
+pytestmark = pytest.mark.anyio
+
+
+def test_otel_span_creates_span():
+    _get_tracer.cache_clear()
+    with otel_span("test.span", kind="CLIENT", attributes={"key": "value"}) as span:
+        assert span is not None
+
+
+def test_otel_span_noop_when_unavailable():
+    _get_tracer.cache_clear()
+    with patch.dict("sys.modules", {"opentelemetry": None, "opentelemetry.trace": None}):
+        _get_tracer.cache_clear()
+        with otel_span("test.span") as span:
+            assert span is None
+    _get_tracer.cache_clear()