feat: Server.run drives JSONRPCDispatcher + ServerRunner; ServerSession is a dispatcher proxy

The swap. Production server traffic now flows through the dispatcher/runner
path; BaseSession is no longer reached on the server side.

ServerRequestContext: standalone dataclass (drops RequestContext base,
inlines session/request_id/meta). ServerMiddleware retyped to take it;
_MwLifespanT no longer contravariant while the ctx is the invariant
mutable dataclass (TODO marked).

Connection: client_params holds the full InitializeRequestParams;
client_info / client_capabilities are read-through properties.

ServerSession: rewritten as a connection-scoped proxy over
JSONRPCDispatcher + Connection. send_request / send_notification
model-dump and forward to dispatcher.send_raw_request / notify, threading
related_request_id so SHTTP routing is unchanged. The typed helpers
(create_message, elicit_*, send_log_message, send_*_list_changed,
list_roots, send_ping, send_progress_notification, send_elicit_complete,
check_client_capability) are kept verbatim. Deleted: the
BaseSession-derived receive loop, _received_*, incoming_messages,
InitializationState, ServerRequestResponder, send_message, the
tasks-only _build_* helpers.

ServerRunner: dispatcher is JSONRPCDispatcher concretely (the
ServerSession shim needs its _related_request_id kwarg). __post_init__
builds the connection-scoped session. _make_context builds
ServerRequestContext from dctx.request_id and dctx.message_metadata
(the same isinstance(ServerMessageMetadata) narrow the previous
Server._handle_request did). _handle_initialize sets
connection.client_params. Both cast(Any, entry.handler) and the
getattr(typed_params, 'meta', ...) are gone (meta read via
isinstance(typed_params, RequestParams)). Server import is under
TYPE_CHECKING to break the cycle with lowlevel/server.

Server.run(): builds JSONRPCDispatcher(read, write,
raise_handler_exceptions=...) and ServerRunner(...,
dispatch_middleware=[otel_middleware]) inside the lifespan, then awaits
runner.run(). _handle_message / _handle_request / _handle_notification
deleted.

Author: maxisbey

src/mcp/server/connection.py

@@ -24,7 +24,7 @@
 from mcp.shared.dispatcher import CallOptions, Outbound
 from mcp.shared.exceptions import NoBackChannelError
 from mcp.shared.peer import Meta, dump_params
-from mcp.types import ClientCapabilities, Implementation, LoggingLevel
+from mcp.types import ClientCapabilities, Implementation, InitializeRequestParams, LoggingLevel
 
 __all__ = ["Connection"]
 
@@ -51,8 +51,8 @@ def __init__(self, outbound: Outbound, *, has_standalone_channel: bool, session_
         self.has_standalone_channel = has_standalone_channel
         self.session_id: str | None = session_id
 
-        self.client_info: Implementation | None = None
-        self.client_capabilities: ClientCapabilities | None = None
+        self.client_params: InitializeRequestParams | None = None
+        """The full `initialize` request params; `None` before initialization."""
         self.protocol_version: str | None = None
         self.initialized: anyio.Event = anyio.Event()
 
@@ -68,6 +68,16 @@ def __init__(self, outbound: Outbound, *, has_standalone_channel: bool, session_
         middleware to register per-connection teardown. Unwound LIFO after
         `dispatcher.run()` returns, shielded from cancellation."""
 
+    @property
+    def client_info(self) -> Implementation | None:
+        """The client's `Implementation` from `initialize`; `None` before initialization."""
+        return self.client_params.client_info if self.client_params is not None else None
+
+    @property
+    def client_capabilities(self) -> ClientCapabilities | None:
+        """The client's `ClientCapabilities` from `initialize`; `None` before initialization."""
+        return self.client_params.capabilities if self.client_params is not None else None
+
     async def send_raw_request(
         self,
         method: str,

src/mcp/server/context.py

@@ -9,24 +9,32 @@
 
 from mcp.server._typed_request import TypedServerRequestMixin
 from mcp.server.connection import Connection
-from mcp.server.experimental.request_context import Experimental
 from mcp.server.session import ServerSession
-from mcp.shared._context import RequestContext
 from mcp.shared.context import BaseContext
 from mcp.shared.dispatcher import DispatchContext
 from mcp.shared.message import CloseSSEStreamCallback
 from mcp.shared.peer import Meta, PeerMixin
 from mcp.shared.transport_context import TransportContext
-from mcp.types import LoggingLevel, RequestParamsMeta
+from mcp.types import LoggingLevel, RequestId, RequestParamsMeta
 
 LifespanContextT = TypeVar("LifespanContextT", default=dict[str, Any])
 RequestT = TypeVar("RequestT", default=Any)
 
 
 @dataclass(kw_only=True)
-class ServerRequestContext(RequestContext[ServerSession], Generic[LifespanContextT, RequestT]):
+class ServerRequestContext(Generic[LifespanContextT, RequestT]):
+    """Per-request context handed to lowlevel request and notification handlers.
+
+    Built by `ServerRunner._make_context` for each inbound message. Carries the
+    connection-scoped `ServerSession` (server-to-client requests and
+    notifications), per-request metadata, and any per-message data the
+    transport attached (the HTTP request, SSE stream-close callbacks).
+    """
+
+    session: ServerSession
     lifespan_context: LifespanContextT
-    experimental: Experimental
+    request_id: RequestId | None = None
+    meta: RequestParamsMeta | None = None
     request: RequestT | None = None
     close_sse_stream: CloseSSEStreamCallback | None = None
     close_standalone_sse_stream: CloseSSEStreamCallback | None = None
@@ -107,26 +115,32 @@ async def log(self, level: LoggingLevel, data: Any, logger: str | None = None, *
 
 CallNext = Callable[[], Awaitable[HandlerResult]]
 
-_MwLifespanT = TypeVar("_MwLifespanT", contravariant=True)
+_MwLifespanT = TypeVar("_MwLifespanT")
 
 
 class ServerMiddleware(Protocol[_MwLifespanT]):
     """Context-tier middleware: `(ctx, method, typed_params, call_next) -> result`.
 
     Runs *inside* `ServerRunner._on_request` after params validation and
-    `Context` construction. Wraps registered handlers (including `ping`) but
+    context construction. Wraps registered handlers (including `ping`) but
     not `initialize`, `METHOD_NOT_FOUND`, or validation failures. Listed
     outermost-first on `Server.middleware`.
 
     `Server[L].middleware` holds `ServerMiddleware[L]`, so an app-specific
-    middleware sees `ctx.lifespan: L`. A reusable middleware can be typed
-    `ServerMiddleware[object]` - `Context` is covariant in `LifespanT`, so it
-    registers on any `Server[L]`.
+    middleware sees `ctx.lifespan_context: L`. While the context is the
+    mutable `ServerRequestContext` dataclass it is invariant in `L`, so a
+    reusable middleware should be typed `ServerMiddleware[Any]` to register on
+    any `Server[L]`.
     """
 
+    # TODO(maxisbey): once `_make_context` returns the (covariant) `Context[L]`
+    # again, restore `_MwLifespanT` to `contravariant=True` and retype `ctx`
+    # below to `Context[_MwLifespanT]` so reusable middleware can be
+    # `ServerMiddleware[object]` instead of `ServerMiddleware[Any]`.
+
     async def __call__(
         self,
-        ctx: Context[_MwLifespanT],
+        ctx: ServerRequestContext[_MwLifespanT, Any],
         method: str,
         params: BaseModel,
         call_next: CallNext,

src/mcp/server/lowlevel/server.py

@@ -36,17 +36,13 @@ async def main():
 
 from __future__ import annotations
 
-import contextvars
 import logging
-import warnings
 from collections.abc import AsyncIterator, Awaitable, Callable
-from contextlib import AbstractAsyncContextManager, AsyncExitStack, asynccontextmanager
+from contextlib import AbstractAsyncContextManager, asynccontextmanager
 from dataclasses import dataclass
 from importlib.metadata import version as importlib_version
-from typing import Any, Generic, cast
+from typing import Any, Generic
 
-import anyio
-from opentelemetry.trace import SpanKind, StatusCode
 from pydantic import BaseModel
 from starlette.applications import Starlette
 from starlette.middleware import Middleware
@@ -61,18 +57,16 @@ async def main():
 from mcp.server.auth.routes import build_resource_metadata_url, create_auth_routes, create_protected_resource_routes
 from mcp.server.auth.settings import AuthSettings
 from mcp.server.context import HandlerResult, ServerMiddleware, ServerRequestContext
-from mcp.server.experimental.request_context import Experimental
 from mcp.server.lowlevel.experimental import ExperimentalHandlers
 from mcp.server.models import InitializationOptions
-from mcp.server.session import ServerSession
+from mcp.server.runner import ServerRunner, otel_middleware
 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
-from mcp.shared.session import RequestResponder
+from mcp.shared.jsonrpc_dispatcher import JSONRPCDispatcher
+from mcp.shared.message import SessionMessage
+from mcp.shared.transport_context import TransportContext
 
 logger = logging.getLogger(__name__)
 
@@ -432,196 +426,23 @@ async def run(
         # the initialization lifecycle, but can do so with any available node
         # rather than requiring initialization for each connection.
         stateless: bool = False,
-    ):
-        async with AsyncExitStack() as stack:
-            lifespan_context = await stack.enter_async_context(self.lifespan(self))
-            session = await stack.enter_async_context(
-                ServerSession(
-                    read_stream,
-                    write_stream,
-                    initialization_options,
-                    stateless=stateless,
-                )
-            )
-
-            # Configure task support for this session if enabled
-            task_support = self._experimental_handlers.task_support if self._experimental_handlers else None
-            if task_support is not None:
-                task_support.configure_session(session)
-                await stack.enter_async_context(task_support.run())
-
-            async with anyio.create_task_group() as tg:
-                try:
-                    async for message in session.incoming_messages:
-                        logger.debug("Received message: %s", message)
-
-                        if isinstance(message, RequestResponder) and message.context is not None:
-                            context = message.context
-                        else:
-                            context = contextvars.copy_context()
-
-                        context.run(
-                            tg.start_soon,
-                            self._handle_message,
-                            message,
-                            session,
-                            lifespan_context,
-                            raise_exceptions,
-                        )
-                finally:
-                    # Transport closed: cancel in-flight handlers. Without this the
-                    # TG join waits for them, and when they eventually try to
-                    # respond they hit a closed write stream (the session's
-                    # _receive_loop closed it when the read stream ended).
-                    tg.cancel_scope.cancel()
-
-    async def _handle_message(
-        self,
-        message: RequestResponder[types.ClientRequest, types.ServerResult] | types.ClientNotification | Exception,
-        session: ServerSession,
-        lifespan_context: LifespanResultT,
-        raise_exceptions: bool = False,
-    ):
-        with warnings.catch_warnings(record=True) as w:
-            match message:
-                case RequestResponder() as responder:
-                    with responder:
-                        await self._handle_request(
-                            message, responder.request, session, lifespan_context, raise_exceptions
-                        )
-                case Exception():
-                    logger.error(f"Received exception from stream: {message}")
-                    if raise_exceptions:
-                        raise message
-                case _:
-                    await self._handle_notification(message, session, lifespan_context)
-
-            for warning in w:  # pragma: lax no cover
-                logger.info("Warning: %s: %s", warning.category.__name__, warning.message)
-
-    async def _handle_request(
-        self,
-        message: RequestResponder[types.ClientRequest, types.ServerResult],
-        req: types.ClientRequest,
-        session: ServerSession,
-        lifespan_context: LifespanResultT,
-        raise_exceptions: bool,
-    ):
-        logger.info("Processing 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}"
-
-        # 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
-
-        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 entry := self._request_handlers.get(req.method):
-                handler = entry.handler
-                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:
-                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:
-                # TODO: cast goes away when `_handle_request` is deleted.
-                await message.respond(cast(types.ServerResult | types.ErrorData, 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,
-        notify: types.ClientNotification,
-        session: ServerSession,
-        lifespan_context: LifespanResultT,
     ) -> None:
-        if entry := self._notification_handlers.get(notify.method):
-            handler = entry.handler
-            logger.debug("Dispatching notification of type %s", type(notify).__name__)
-
-            try:
-                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
-                ctx = ServerRequestContext(
-                    session=session,
-                    lifespan_context=lifespan_context,
-                    experimental=Experimental(
-                        task_metadata=None,
-                        _client_capabilities=client_capabilities,
-                        _session=session,
-                        _task_support=task_support,
-                    ),
-                )
-                await handler(ctx, notify.params)
-            except Exception:  # pragma: no cover
-                logger.exception("Uncaught exception in notification handler")
+        async with self.lifespan(self) as lifespan_context:
+            dispatcher: JSONRPCDispatcher[TransportContext] = JSONRPCDispatcher(
+                read_stream,
+                write_stream,
+                raise_handler_exceptions=raise_exceptions,
+            )
+            runner = ServerRunner(
+                server=self,
+                dispatcher=dispatcher,
+                lifespan_state=lifespan_context,
+                init_options=initialization_options,
+                has_standalone_channel=True,
+                stateless=stateless,
+                dispatch_middleware=[otel_middleware],
+            )
+            await runner.run()
 
     def streamable_http_app(
         self,