feat: ServerRunner.run() and otel_middleware

run() composes dispatch_middleware over _on_request and forwards task_status
to dispatcher.run() so callers can 'await tg.start(runner.run)'.

otel_middleware is a DispatchMiddleware that wraps each request in a span,
mirroring the existing Server._handle_request span shape: name 'MCP handle
<method> [<target>]', mcp.method.name attribute, W3C trace context extracted
from params._meta (SEP-414), and ERROR status if the handler raises.

connection_lifespan plumbing (the enter-late dance) is deferred to a separate
commit since Server.connection_lifespan is None today.

Author: maxisbey

src/mcp/server/runner.py

@@ -22,12 +22,15 @@
 from functools import partial, reduce
 from typing import Any, Generic, Protocol, cast
 
+import anyio.abc
+from opentelemetry.trace import SpanKind, StatusCode
 from pydantic import BaseModel
 from typing_extensions import TypeVar
 
 from mcp.server.connection import Connection
 from mcp.server.context import CallNext, Context, ContextMiddleware
 from mcp.server.lowlevel.server import NotificationOptions
+from mcp.shared._otel import extract_trace_context, otel_span
 from mcp.shared.dispatcher import DispatchContext, Dispatcher, DispatchMiddleware, OnRequest
 from mcp.shared.exceptions import MCPError
 from mcp.shared.transport_context import TransportContext
@@ -52,7 +55,7 @@
     UnsubscribeRequestParams,
 )
 
-__all__ = ["CallNext", "ContextMiddleware", "ServerRegistry", "ServerRunner"]
+__all__ = ["CallNext", "ContextMiddleware", "ServerRegistry", "ServerRunner", "otel_middleware"]
 
 logger = logging.getLogger(__name__)
 
@@ -117,6 +120,44 @@ def get_capabilities(
     ) -> ServerCapabilities: ...
 
 
+def otel_middleware(next_on_request: OnRequest) -> OnRequest:
+    """Dispatch-tier middleware that wraps each request in an OpenTelemetry span.
+
+    Mirrors the span shape of the existing `Server._handle_request`: span name
+    ``"MCP handle <method> [<target>]"``, ``mcp.method.name`` attribute, W3C
+    trace context extracted from ``params._meta`` (SEP-414), and an ERROR
+    status if the handler raises.
+    """
+
+    async def wrapped(
+        dctx: DispatchContext[TransportContext], method: str, params: Mapping[str, Any] | None
+    ) -> dict[str, Any]:
+        target: str | None
+        match params:
+            case {"name": str() as target}:
+                pass
+            case _:
+                target = None
+        parent: Any | None
+        match params:
+            case {"_meta": {**meta}}:
+                parent = extract_trace_context(meta)
+            case _:
+                parent = None
+        span_name = f"MCP handle {method}{f' {target}' if target else ''}"
+        with otel_span(span_name, kind=SpanKind.SERVER, attributes={"mcp.method.name": method}, context=parent) as span:
+            try:
+                return await next_on_request(dctx, method, params)
+            except MCPError as e:
+                span.set_status(StatusCode.ERROR, e.error.message)
+                raise
+            except Exception as e:
+                span.set_status(StatusCode.ERROR, str(e))
+                raise
+
+    return wrapped
+
+
 def _dump_result(result: Any) -> dict[str, Any]:
     if result is None:
         return {}
@@ -145,6 +186,16 @@ def __post_init__(self) -> None:
         self._initialized = self.stateless
         self.connection = Connection(self.dispatcher, has_standalone_channel=self.has_standalone_channel)
 
+    async def run(self, *, task_status: anyio.abc.TaskStatus[None] = anyio.TASK_STATUS_IGNORED) -> None:
+        """Drive the dispatcher until the underlying channel closes.
+
+        Composes `dispatch_middleware` over `_on_request` and hands the result
+        to `dispatcher.run()`. ``task_status.started()`` is forwarded so callers
+        can ``await tg.start(runner.run)`` and resume once the dispatcher is
+        ready to accept requests.
+        """
+        await self.dispatcher.run(self._compose_on_request(), self._on_notify, task_status=task_status)
+
     def _compose_on_request(self) -> OnRequest:
         """Wrap `_on_request` in `dispatch_middleware`, outermost-first.
 

tests/server/test_runner.py

@@ -14,7 +14,7 @@
 from mcp.server.connection import Connection
 from mcp.server.context import Context
 from mcp.server.lowlevel.server import Server
-from mcp.server.runner import ServerRunner
+from mcp.server.runner import ServerRunner, otel_middleware
 from mcp.shared.direct_dispatcher import create_direct_dispatcher_pair
 from mcp.shared.exceptions import MCPError
 from mcp.shared.transport_context import TransportContext
@@ -272,6 +272,75 @@ async def mw(ctx: Any, method: str, params: Any, call_next: Any) -> Any:
         tg.cancel_scope.cancel()
 
 
+@pytest.mark.anyio
+async def test_runner_run_drives_dispatcher_end_to_end(server: SrvT):
+    client, server_d = create_direct_dispatcher_pair()
+    runner = ServerRunner(server=server, dispatcher=server_d, lifespan_state=None, has_standalone_channel=True)
+    c_req, c_notify = echo_handlers(Recorder())
+    async with anyio.create_task_group() as tg:
+        await tg.start(client.run, c_req, c_notify)
+        await tg.start(runner.run)
+        with anyio.fail_after(5):
+            init = await client.send_raw_request("initialize", _initialize_params())
+            tools = await client.send_raw_request("tools/list", None)
+        assert init["serverInfo"]["name"] == "test-server"
+        assert tools["tools"][0]["name"] == "t"
+        tg.cancel_scope.cancel()
+
+
+@pytest.mark.anyio
+async def test_runner_run_applies_dispatch_middleware(server: SrvT):
+    seen: list[str] = []
+
+    def trace_mw(next_on_request: Any) -> Any:
+        async def wrapped(dctx: Any, method: str, params: Any) -> Any:
+            seen.append(method)
+            return await next_on_request(dctx, method, params)
+
+        return wrapped
+
+    client, server_d = create_direct_dispatcher_pair()
+    runner = ServerRunner(
+        server=server,
+        dispatcher=server_d,
+        lifespan_state=None,
+        has_standalone_channel=True,
+        dispatch_middleware=[trace_mw],
+    )
+    c_req, c_notify = echo_handlers(Recorder())
+    async with anyio.create_task_group() as tg:
+        await tg.start(client.run, c_req, c_notify)
+        await tg.start(runner.run)
+        with anyio.fail_after(5):
+            await client.send_raw_request("initialize", _initialize_params())
+            await client.send_raw_request("ping", None)
+        assert seen == ["initialize", "ping"]
+        tg.cancel_scope.cancel()
+
+
+@pytest.mark.anyio
+async def test_otel_middleware_passes_through_result_and_survives_handler_error(server: SrvT):
+    client, server_d = create_direct_dispatcher_pair()
+    runner = ServerRunner(
+        server=server,
+        dispatcher=server_d,
+        lifespan_state=None,
+        has_standalone_channel=True,
+        dispatch_middleware=[otel_middleware],
+    )
+    c_req, c_notify = echo_handlers(Recorder())
+    async with anyio.create_task_group() as tg:
+        await tg.start(client.run, c_req, c_notify)
+        await tg.start(runner.run)
+        with anyio.fail_after(5):
+            await client.send_raw_request("initialize", _initialize_params())
+            tools = await client.send_raw_request("tools/list", None)
+            assert tools["tools"][0]["name"] == "t"
+            with pytest.raises(MCPError):
+                await client.send_raw_request("nonexistent/method", None)
+        tg.cancel_scope.cancel()
+
+
 @pytest.mark.anyio
 async def test_runner_stateless_skips_init_gate(server: SrvT):
     client, server_d = create_direct_dispatcher_pair()