feat: add instrumentation interface for observability

Add a pluggable instrumentation interface for monitoring MCP request/response lifecycle. This lays groundwork for OpenTelemetry and other observability integrations.

Changes:
- Define Instrumenter protocol with on_request_start, on_request_end, and on_error hooks
- Add NoOpInstrumenter as default implementation with minimal overhead
- Wire instrumenter into ServerSession and ClientSession constructors
- Add instrumentation calls in Server._handle_request for server-side monitoring
- Add request_id to log records via extra field for correlation
- Add comprehensive tests for instrumentation protocol
- Add documentation with examples and best practices

Addresses #421

Author: dgenio

docs/instrumentation.md

@@ -0,0 +1,208 @@
+# Instrumentation
+
+The MCP Python SDK provides a pluggable instrumentation interface for monitoring request/response lifecycle. This enables integration with OpenTelemetry, custom metrics, logging frameworks, and other observability tools.
+
+## Overview
+
+The `Instrumenter` protocol defines three hooks:
+
+- `on_request_start`: Called when a request starts processing
+- `on_request_end`: Called when a request completes (successfully or not)
+- `on_error`: Called when an error occurs during request processing
+
+All methods are optional (no-op implementations are valid). Exceptions raised by instrumentation hooks are logged but do not affect request processing.
+
+## Basic Usage
+
+### Server-Side Instrumentation
+
+```python
+from mcp.server.lowlevel import Server
+from mcp.shared.instrumentation import Instrumenter
+from mcp.types import RequestId
+
+class MyInstrumenter:
+    """Custom instrumenter implementation."""
+    
+    def on_request_start(
+        self,
+        request_id: RequestId,
+        request_type: str,
+        method: str | None = None,
+        **metadata,
+    ) -> None:
+        print(f"Request {request_id} started: {request_type}")
+    
+    def on_request_end(
+        self,
+        request_id: RequestId,
+        request_type: str,
+        success: bool,
+        duration_seconds: float | None = None,
+        **metadata,
+    ) -> None:
+        status = "succeeded" if success else "failed"
+        print(f"Request {request_id} {status} in {duration_seconds:.3f}s")
+    
+    def on_error(
+        self,
+        request_id: RequestId | None,
+        error: Exception,
+        error_type: str,
+        **metadata,
+    ) -> None:
+        print(f"Error in request {request_id}: {error_type} - {error}")
+
+# Create server with custom instrumenter
+server = Server("my-server")
+
+# Pass instrumenter when running the server
+async def run_server():
+    async with stdio_server() as (read_stream, write_stream):
+        await server.run(
+            read_stream,
+            write_stream,
+            server.create_initialization_options(),
+            instrumenter=MyInstrumenter(),
+        )
+```
+
+### Client-Side Instrumentation
+
+```python
+from mcp.client.session import ClientSession
+from mcp.shared.instrumentation import Instrumenter
+
+# Create client session with instrumenter
+async with ClientSession(
+    read_stream=read_stream,
+    write_stream=write_stream,
+    instrumenter=MyInstrumenter(),
+) as session:
+    await session.initialize()
+    # Use session...
+```
+
+## Metadata
+
+Instrumentation hooks receive metadata via `**metadata` keyword arguments:
+
+- `on_request_start` metadata:
+  - `session_type`: "server" or "client"
+  - Any additional context provided by the framework
+
+- `on_request_end` metadata:
+  - `cancelled`: True if the request was cancelled
+  - `error`: Error message if request failed
+  - Any additional context
+
+- `on_error` metadata:
+  - Additional error context
+
+## Request ID
+
+The `request_id` parameter is consistent across all hooks for a given request, allowing you to correlate the request lifecycle. The `request_id` is also added to log records via the `extra` field, so you can filter logs by request.
+
+## OpenTelemetry Integration
+
+A full OpenTelemetry instrumenter will be provided in a future release or as a separate package. Here's a basic example to get started:
+
+```python
+from opentelemetry import trace
+from opentelemetry.trace import Status, StatusCode
+
+tracer = trace.get_tracer(__name__)
+
+class OpenTelemetryInstrumenter:
+    def __init__(self):
+        self.spans = {}
+    
+    def on_request_start(self, request_id, request_type, **metadata):
+        span = tracer.start_span(
+            f"mcp.request.{request_type}",
+            attributes={
+                "mcp.request_id": str(request_id),
+                "mcp.request_type": request_type,
+                **metadata,
+            }
+        )
+        self.spans[request_id] = span
+    
+    def on_request_end(self, request_id, request_type, success, duration_seconds=None, **metadata):
+        if span := self.spans.pop(request_id, None):
+            if duration_seconds:
+                span.set_attribute("mcp.duration_seconds", duration_seconds)
+            span.set_status(Status(StatusCode.OK if success else StatusCode.ERROR))
+            span.end()
+    
+    def on_error(self, request_id, error, error_type, **metadata):
+        if span := self.spans.get(request_id):
+            span.record_exception(error)
+            span.set_status(Status(StatusCode.ERROR, str(error)))
+```
+
+## Default Behavior
+
+If no instrumenter is provided, a no-op implementation is used automatically. This has minimal overhead and doesn't affect request processing.
+
+```python
+from mcp.shared.instrumentation import get_default_instrumenter
+
+# Get the default no-op instrumenter
+instrumenter = get_default_instrumenter()
+```
+
+## Best Practices
+
+1. **Keep hooks fast**: Instrumentation hooks are called synchronously in the request path. Keep processing minimal to avoid impacting request latency.
+
+2. **Handle errors gracefully**: Exceptions in instrumentation hooks are caught and logged, but it's best to handle errors within your instrumenter.
+
+3. **Use appropriate metadata**: Include relevant context in metadata fields to aid debugging and analysis.
+
+4. **Consider sampling**: For high-volume servers, consider implementing sampling in your instrumenter to reduce overhead.
+
+## Example: Custom Metrics
+
+```python
+from collections import defaultdict
+from typing import Dict
+
+class MetricsInstrumenter:
+    """Track request counts and durations."""
+    
+    def __init__(self):
+        self.request_counts: Dict[str, int] = defaultdict(int)
+        self.request_durations: Dict[str, list[float]] = defaultdict(list)
+        self.error_counts: Dict[str, int] = defaultdict(int)
+    
+    def on_request_start(self, request_id, request_type, **metadata):
+        self.request_counts[request_type] += 1
+    
+    def on_request_end(self, request_id, request_type, success, duration_seconds=None, **metadata):
+        if duration_seconds is not None:
+            self.request_durations[request_type].append(duration_seconds)
+    
+    def on_error(self, request_id, error, error_type, **metadata):
+        self.error_counts[error_type] += 1
+    
+    def get_stats(self):
+        """Get statistics summary."""
+        stats = {}
+        for request_type, durations in self.request_durations.items():
+            if durations:
+                avg_duration = sum(durations) / len(durations)
+                stats[request_type] = {
+                    "count": self.request_counts[request_type],
+                    "avg_duration": avg_duration,
+                }
+        return stats
+```
+
+## Future Work
+
+- Full OpenTelemetry integration as a separate module
+- Additional built-in instrumenters (Prometheus, StatsD, etc.)
+- Client-side request instrumentation
+- Async hook support for long-running instrumentation operations
+

mkdocs.yml

@@ -18,6 +18,7 @@ nav:
       - Low-Level Server: low-level-server.md
       - Authorization: authorization.md
       - Testing: testing.md
+      - Instrumentation: instrumentation.md
   - API Reference: api.md
 
 theme:

src/mcp/client/session.py

@@ -9,6 +9,7 @@
 
 import mcp.types as types
 from mcp.shared.context import RequestContext
+from mcp.shared.instrumentation import Instrumenter, get_default_instrumenter
 from mcp.shared.message import SessionMessage
 from mcp.shared.session import BaseSession, ProgressFnT, RequestResponder
 from mcp.shared.version import SUPPORTED_PROTOCOL_VERSIONS
@@ -118,6 +119,7 @@ def __init__(
         logging_callback: LoggingFnT | None = None,
         message_handler: MessageHandlerFnT | None = None,
         client_info: types.Implementation | None = None,
+        instrumenter: Instrumenter | None = None,
     ) -> None:
         super().__init__(
             read_stream,
@@ -127,6 +129,7 @@ def __init__(
             read_timeout_seconds=read_timeout_seconds,
         )
         self._client_info = client_info or DEFAULT_CLIENT_INFO
+        self._instrumenter = instrumenter or get_default_instrumenter()
         self._sampling_callback = sampling_callback or _default_sampling_callback
         self._elicitation_callback = elicitation_callback or _default_elicitation_callback
         self._list_roots_callback = list_roots_callback or _default_list_roots_callback
@@ -135,6 +138,11 @@ def __init__(
         self._tool_output_schemas: dict[str, dict[str, Any] | None] = {}
         self._server_capabilities: types.ServerCapabilities | None = None
 
+    @property
+    def instrumenter(self) -> Instrumenter:
+        """Get the instrumenter for this session."""
+        return self._instrumenter
+
     async def initialize(self) -> types.InitializeResult:
         sampling = types.SamplingCapability() if self._sampling_callback is not _default_sampling_callback else None
         elicitation = (

src/mcp/server/lowlevel/server.py

@@ -70,6 +70,7 @@ async def main():
 import contextvars
 import json
 import logging
+import time
 import warnings
 from collections.abc import AsyncIterator, Awaitable, Callable, Iterable
 from contextlib import AbstractAsyncContextManager, AsyncExitStack, asynccontextmanager
@@ -85,6 +86,7 @@ async def main():
 from mcp.server.lowlevel.func_inspection import create_call_wrapper
 from mcp.server.lowlevel.helper_types import ReadResourceContents
 from mcp.server.models import InitializationOptions
+from mcp.shared.instrumentation import Instrumenter
 from mcp.server.session import ServerSession
 from mcp.shared.context import RequestContext
 from mcp.shared.exceptions import McpError
@@ -615,6 +617,7 @@ async def run(
         # the initialization lifecycle, but can do so with any available node
         # rather than requiring initialization for each connection.
         stateless: bool = False,
+        instrumenter: Instrumenter | None = None,
     ):
         async with AsyncExitStack() as stack:
             lifespan_context = await stack.enter_async_context(self.lifespan(self))
@@ -624,6 +627,7 @@ async def run(
                     write_stream,
                     initialization_options,
                     stateless=stateless,
+                    instrumenter=instrumenter,
                 )
             )
 
@@ -674,11 +678,27 @@ async def _handle_request(
         lifespan_context: LifespanResultT,
         raise_exceptions: bool,
     ):
-        logger.info("Processing request of type %s", type(req).__name__)
+        request_type = type(req).__name__
+        log_extra = {"request_id": str(message.request_id)}
+        logger.info("Processing request of type %s", request_type, extra=log_extra)
+        
+        # Start instrumentation
+        start_time = time.monotonic()
+        try:
+            session.instrumenter.on_request_start(
+                request_id=message.request_id,
+                request_type=request_type,
+                session_type="server",
+            )
+        except Exception:  # pragma: no cover
+            logger.exception("Error in instrumentation on_request_start")
+
         if handler := self.request_handlers.get(type(req)):  # type: ignore
-            logger.debug("Dispatching request of type %s", type(req).__name__)
+            logger.debug("Dispatching request of type %s", request_type, extra=log_extra)
 
             token = None
+            response = None
+            success = False
             try:
                 # Extract request context from message metadata
                 request_data = None
@@ -699,22 +719,61 @@ async def _handle_request(
                     )
                 )
                 response = await handler(req)
+                success = not isinstance(response, types.ErrorData)
             except McpError as err:  # pragma: no cover
                 response = err.error
+                try:
+                    session.instrumenter.on_error(
+                        request_id=message.request_id,
+                        error=err,
+                        error_type=type(err).__name__,
+                    )
+                except Exception:  # pragma: no cover
+                    logger.exception("Error in instrumentation on_error")
             except anyio.get_cancelled_exc_class():  # pragma: no cover
                 logger.info(
                     "Request %s cancelled - duplicate response suppressed",
                     message.request_id,
+                    extra=log_extra,
                 )
+                try:
+                    session.instrumenter.on_request_end(
+                        request_id=message.request_id,
+                        request_type=request_type,
+                        success=False,
+                        duration_seconds=time.monotonic() - start_time,
+                        cancelled=True,
+                    )
+                except Exception:  # pragma: no cover
+                    logger.exception("Error in instrumentation on_request_end")
                 return
             except Exception as err:  # pragma: no cover
+                try:
+                    session.instrumenter.on_error(
+                        request_id=message.request_id,
+                        error=err,
+                        error_type=type(err).__name__,
+                    )
+                except Exception:  # pragma: no cover
+                    logger.exception("Error in instrumentation on_error")
                 if raise_exceptions:
                     raise err
                 response = types.ErrorData(code=0, message=str(err), data=None)
             finally:
                 # Reset the global state after we are done
                 if token is not None:  # pragma: no branch
                     request_ctx.reset(token)
+                
+                # End instrumentation
+                try:
+                    session.instrumenter.on_request_end(
+                        request_id=message.request_id,
+                        request_type=request_type,
+                        success=success,
+                        duration_seconds=time.monotonic() - start_time,
+                    )
+                except Exception:  # pragma: no cover
+                    logger.exception("Error in instrumentation on_request_end")
 
             await message.respond(response)
         else:  # pragma: no cover
@@ -724,8 +783,18 @@ async def _handle_request(
                     message="Method not found",
                 )
             )
+            try:
+                session.instrumenter.on_request_end(
+                    request_id=message.request_id,
+                    request_type=request_type,
+                    success=False,
+                    duration_seconds=time.monotonic() - start_time,
+                    error="Method not found",
+                )
+            except Exception:  # pragma: no cover
+                logger.exception("Error in instrumentation on_request_end")
 
-        logger.debug("Response sent")
+        logger.debug("Response sent", extra=log_extra)
 
     async def _handle_notification(self, notify: Any):
         if handler := self.notification_handlers.get(type(notify)):  # type: ignore