feat: implement mcp.server.operation.duration and mcp.server.session.duration metrics

Implements the two server-side OTel metrics defined in the MCP semantic
conventions (https://opentelemetry.io/docs/specs/semconv/gen-ai/mcp), as
discussed in #421.

Changes:
- src/mcp/shared/_otel.py: add a meter and two histograms with the
  spec-mandated explicit bucket boundaries
  [0.01, 0.02, 0.05, 0.1, 0.2, 0.5, 1, 2, 5, 10, 30, 60, 120, 300].
  Expose record_server_operation_duration() and
  record_server_session_duration() helpers.
- src/mcp/server/lowlevel/server.py: record mcp.server.operation.duration
  in _handle_request() on all exit paths (success, MCPError, cancellation,
  transport close, raise_exceptions=True). Attributes recorded:
  mcp.method.name (required), error.type, rpc.response.status_code,
  gen_ai.tool.name + gen_ai.operation.name (tools/call),
  gen_ai.prompt.name (prompts/get), mcp.protocol.version.
- src/mcp/server/session.py: record mcp.server.session.duration in
  __aexit__(). Session start time is captured in __aenter__() (not
  __init__()) so the clock starts when the session is actually active.
  Cancellation exceptions (transport close) are not treated as errors.
- tests/shared/test_otel.py: extend test_client_and_server_instrumentation
  to assert both metrics are emitted with correct attributes and units.

Transport attributes (network.transport, network.protocol.name,
network.protocol.version) are not set. ServerSession does not currently
know the transport type — adding it would require a new constructor
parameter and a corresponding API change. This can be addressed in a
follow-up once the transport kind is plumbed through.

Generated by Mistral Vibe.
Co-Authored-By: Mistral Vibe <vibe@mistral.ai>

Author: verdie-g

src/mcp/server/lowlevel/server.py

@@ -38,6 +38,7 @@ async def main():
 
 import contextvars
 import logging
+import time
 import warnings
 from collections.abc import AsyncIterator, Awaitable, Callable
 from contextlib import AbstractAsyncContextManager, AsyncExitStack, asynccontextmanager
@@ -66,7 +67,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._otel import extract_trace_context, otel_span, record_server_operation_duration
 from mcp.shared._stream_protocols import ReadStream, WriteStream
 from mcp.shared.exceptions import MCPError
 from mcp.shared.message import ServerMessageMetadata, SessionMessage
@@ -455,12 +456,34 @@ async def _handle_request(
         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
 
+        mcp_protocol_version: str | None = session.client_params.protocol_version if session.client_params else None
+
+        start_time = time.monotonic()
+
+        def _record_duration(
+            *,
+            error_type: str | None = None,
+            rpc_response_status_code: str | None = None,
+        ) -> None:
+            record_server_operation_duration(
+                time.monotonic() - start_time,
+                req.method,
+                error_type=error_type,
+                rpc_response_status_code=rpc_response_status_code,
+                tool_name=target if req.method == "tools/call" else None,
+                prompt_name=target if req.method == "prompts/get" else None,
+                mcp_protocol_version=mcp_protocol_version,
+            )
+
         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:
+            error_type: str | None = None
+            rpc_response_status_code: str | None = None
+
             if handler := self._request_handlers.get(req.method):
                 logger.debug("Dispatching request of type %s", type(req).__name__)
 
@@ -499,25 +522,38 @@ async def _handle_request(
                     )
                     response = await handler(ctx, req.params)
                 except MCPError as err:
+                    rpc_response_status_code = str(err.error.code)
+                    error_type = rpc_response_status_code
                     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)
+                        _record_duration(error_type="cancelled")
                         return
                     # Transport-close cancellation from the TG in run(); re-raise so the
                     # TG swallows its own cancellation.
                     raise
                 except Exception as err:
+                    error_type = type(err).__name__
                     if raise_exceptions:  # pragma: no cover
+                        _record_duration(error_type=error_type)
                         raise err
                     response = types.ErrorData(code=0, message=str(err))
             else:  # pragma: no cover
+                rpc_response_status_code = str(types.METHOD_NOT_FOUND)
+                error_type = rpc_response_status_code
                 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)
+            if isinstance(response, types.ErrorData):
+                if span is not None:
+                    span.set_status(StatusCode.ERROR, response.message)
+                # Only set error_type/rpc_response_status_code from response code if not
+                # already set by an exception.
+                if error_type is None:
+                    rpc_response_status_code = str(response.code)
+                    error_type = rpc_response_status_code
 
             try:
                 await message.respond(response)
@@ -529,10 +565,13 @@ async def _handle_request(
                 # 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)
+                _record_duration(error_type=error_type, rpc_response_status_code=rpc_response_status_code)
                 return
 
             logger.debug("Response sent")
 
+        _record_duration(error_type=error_type, rpc_response_status_code=rpc_response_status_code)
+
     async def _handle_notification(
         self,
         notify: types.ClientNotification,

src/mcp/server/session.py

@@ -40,6 +40,9 @@ async def handle_list_prompts(ctx: RequestContext, params) -> ListPromptsResult:
 from mcp.server.experimental.session_features import ExperimentalServerSessionFeatures
 from mcp.server.models import InitializationOptions
 from mcp.server.validation import validate_sampling_tools, validate_tool_use_result_messages
+import time
+
+from mcp.shared._otel import record_server_session_duration
 from mcp.shared._stream_protocols import ReadStream, WriteStream
 from mcp.shared.exceptions import StatelessModeNotSupported
 from mcp.shared.experimental.tasks.capabilities import check_tasks_capability
@@ -96,6 +99,27 @@ def __init__(
             ServerRequestResponder
         ](0)
         self._exit_stack.push_async_callback(lambda: self._incoming_message_stream_reader.aclose())
+        self._session_start_time: float | None = None
+
+    async def __aenter__(self) -> "ServerSession":
+        self._session_start_time = time.monotonic()
+        return await super().__aenter__()
+
+    async def __aexit__(self, exc_type: type[BaseException] | None, exc_val: BaseException | None, exc_tb: Any) -> bool | None:
+        if self._session_start_time is not None:
+            duration = time.monotonic() - self._session_start_time
+            mcp_protocol_version: str | None = (
+                self._client_params.protocol_version if self._client_params else None
+            )
+            # Cancellation exceptions indicate transport close, not a session error.
+            is_cancellation = exc_val is not None and isinstance(exc_val, anyio.get_cancelled_exc_class())
+            error_type: str | None = type(exc_val).__name__ if exc_val is not None and not is_cancellation else None
+            record_server_session_duration(
+                duration,
+                error_type=error_type,
+                mcp_protocol_version=mcp_protocol_version,
+            )
+        return await super().__aexit__(exc_type, exc_val, exc_tb)
 
     @property
     def _receive_request_adapter(self) -> TypeAdapter[types.ClientRequest]:

src/mcp/shared/_otel.py

@@ -7,10 +7,32 @@
 from typing import Any
 
 from opentelemetry.context import Context
+from opentelemetry.metrics import get_meter
 from opentelemetry.propagate import extract, inject
 from opentelemetry.trace import SpanKind, get_tracer
 
 _tracer = get_tracer("mcp-python-sdk")
+_meter = get_meter("mcp-python-sdk")
+
+# Metrics as defined by the OTEL semconv https://github.com/open-telemetry/semantic-conventions/blob/main/docs/gen-ai/mcp.md
+_DURATION_BUCKETS = [0.01, 0.02, 0.05, 0.1, 0.2, 0.5, 1, 2, 5, 10, 30, 60, 120, 300]
+
+_server_operation_duration = _meter.create_histogram(
+    "mcp.server.operation.duration",
+    unit="s",
+    description=(
+        "MCP request or notification duration as observed on the receiver "
+        "from the time it was received until the result or ack is sent."
+    ),
+    explicit_bucket_boundaries_advisory=_DURATION_BUCKETS,
+)
+
+_server_session_duration = _meter.create_histogram(
+    "mcp.server.session.duration",
+    unit="s",
+    description="The duration of the MCP session as observed on the MCP server.",
+    explicit_bucket_boundaries_advisory=_DURATION_BUCKETS,
+)
 
 
 @contextmanager
@@ -34,3 +56,44 @@ def inject_trace_context(meta: dict[str, Any]) -> None:
 def extract_trace_context(meta: dict[str, Any]) -> Context:
     """Extract W3C trace context from a `_meta` dict."""
     return extract(meta)
+
+
+def record_server_operation_duration(
+    duration_s: float,
+    method: str,
+    *,
+    error_type: str | None = None,
+    rpc_response_status_code: str | None = None,
+    tool_name: str | None = None,
+    prompt_name: str | None = None,
+    mcp_protocol_version: str | None = None,
+) -> None:
+    """Record a data point for mcp.server.operation.duration."""
+    attributes: dict[str, str] = {"mcp.method.name": method}
+    if error_type is not None:
+        attributes["error.type"] = error_type
+    if rpc_response_status_code is not None:
+        attributes["rpc.response.status_code"] = rpc_response_status_code
+    if tool_name is not None:
+        attributes["gen_ai.tool.name"] = tool_name
+        attributes["gen_ai.operation.name"] = "execute_tool"
+    if prompt_name is not None:
+        attributes["gen_ai.prompt.name"] = prompt_name
+    if mcp_protocol_version is not None:
+        attributes["mcp.protocol.version"] = mcp_protocol_version
+    _server_operation_duration.record(duration_s, attributes)
+
+
+def record_server_session_duration(
+    duration_s: float,
+    *,
+    error_type: str | None = None,
+    mcp_protocol_version: str | None = None,
+) -> None:
+    """Record a data point for mcp.server.session.duration."""
+    attributes: dict[str, str] = {}
+    if error_type is not None:
+        attributes["error.type"] = error_type
+    if mcp_protocol_version is not None:
+        attributes["mcp.protocol.version"] = mcp_protocol_version
+    _server_session_duration.record(duration_s, attributes)

tests/shared/test_otel.py

@@ -13,8 +13,8 @@
 # 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."""
+async def test_client_and_server_instrumentation(capfire: CaptureLogfire):
+    """Verify that calling a tool produces client and server spans and metrics with correct attributes."""
     server = MCPServer("test")
 
     @server.tool()
@@ -42,3 +42,30 @@ def greet(name: str) -> str:
 
     # 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"]
+
+    metrics = {m["name"]: m for m in capfire.get_collected_metrics() if m["name"].startswith("mcp.")}
+
+    assert "mcp.server.operation.duration" in metrics
+    assert "mcp.server.session.duration" in metrics
+
+    op_metric = metrics["mcp.server.operation.duration"]
+    assert op_metric["unit"] == "s"
+    op_points = op_metric["data"]["data_points"]
+
+    # tools/call data point
+    tools_call_point = next(p for p in op_points if p["attributes"]["mcp.method.name"] == "tools/call")
+    assert tools_call_point["attributes"]["gen_ai.tool.name"] == "greet"
+    assert tools_call_point["attributes"]["gen_ai.operation.name"] == "execute_tool"
+    assert tools_call_point["attributes"]["mcp.protocol.version"] == "2025-11-25"
+    assert tools_call_point["count"] == 1
+    assert tools_call_point["sum"] > 0
+
+    # tools/list is also called during initialization
+    assert any(p["attributes"]["mcp.method.name"] == "tools/list" for p in op_points)
+
+    session_metric = metrics["mcp.server.session.duration"]
+    assert session_metric["unit"] == "s"
+    [session_point] = session_metric["data"]["data_points"]
+    assert session_point["attributes"]["mcp.protocol.version"] == "2025-11-25"
+    assert session_point["count"] == 1
+    assert session_point["sum"] > 0