S2 wave 4: modern entry rewrite, Server.run wrapper, test migrations

Modern HTTP entry (_streamable_http_modern.py):
- handle_modern_request: classify_inbound_request -> on rejection write JSONRPCError at
  ERROR_CODE_HTTP_STATUS-mapped status; on route -> Connection.from_envelope -> serve_one
- All status codes via ERROR_CODE_HTTP_STATUS table (PARSE_ERROR added: 400)
- id:null preserved on parse-error responses (exclude_none would drop it)
- SingleExchangeDispatcher deleted (-73 lines); duplicate shielded-teardown deleted;
  both reportPrivateUsage ignores deleted; per-request lifespan entry deleted

StreamableHTTPSessionManager:
- run() enters lifespan once at startup; lifespan_state stored on self
- _handle_stateless_request rewired: builds JSONRPCDispatcher (can_send_request=False so
  the per-request channel raises NoBackChannelError for server->client requests, allows
  notifications) + Connection.from_envelope(header-seeded version) -> serve_connection.
  Bypasses Server.run(); stateless kwarg removed from both manager call sites.

lowlevel.Server.run():
- Body rewritten: enter lifespan -> JSONRPCDispatcher -> Connection.for_loop ->
  serve_connection. stateless kwarg deleted (no callers). Net -4 LoC.

Test migrations:
- test_runner.py: connected_runner uses factories; resolver tests deleted; per-request
  session tests; serve_one/serve_connection/to_jsonrpc_response/aclose_shielded coverage
- test_connection.py: factory tests; check_capability via from_envelope
- test_session.py: StubOutbound; two-channel selector tests (4 new)
- test_stateless_mode.py: rewritten; 8x pytest.raises(NoBackChannelError) via _NoChannelOutbound
- test_streamable_http_modern.py: classifier-path tests; SingleExchangeDispatcher tests deleted
- test_lowlevel_exception_handling.py: stateless=True kwarg removed
- test_streamable_http.py:1371: protocol_version assertion updated (non-Optional)

Part of #2891.

Author: maxisbey

src/mcp/server/_streamable_http_modern.py

@@ -12,33 +12,32 @@
 
 from __future__ import annotations
 
+import json
 import logging
 from collections.abc import Mapping
 from dataclasses import dataclass, field
-from typing import TYPE_CHECKING, Any
+from typing import TYPE_CHECKING, Any, TypeVar
 
 import anyio
-import anyio.abc
-from pydantic import ValidationError
+from pydantic import BaseModel, ValidationError
 from starlette.requests import Request
 from starlette.responses import Response
 from starlette.types import Receive, Scope, Send
 
-from mcp.server.runner import (
-    _EXIT_STACK_CLOSE_TIMEOUT,  # type: ignore[reportPrivateUsage]
-    ServerRunner,
-    otel_middleware,
-)
+from mcp.server.connection import Connection
+from mcp.server.runner import serve_one
 from mcp.server.transport_security import TransportSecurityMiddleware, TransportSecuritySettings
-from mcp.shared.dispatcher import CallOptions, OnNotify, OnRequest
-from mcp.shared.exceptions import MCPError, NoBackChannelError
+from mcp.shared.dispatcher import CallOptions
+from mcp.shared.exceptions import NoBackChannelError
+from mcp.shared.inbound import ERROR_CODE_HTTP_STATUS, InboundLadderRejection, classify_inbound_request
 from mcp.shared.message import MessageMetadata, ServerMessageMetadata
 from mcp.shared.transport_context import TransportContext
 from mcp.types import (
-    INTERNAL_ERROR,
-    INVALID_PARAMS,
+    METHOD_NOT_FOUND,
     PARSE_ERROR,
+    ClientCapabilities,
     ErrorData,
+    Implementation,
     JSONRPCError,
     JSONRPCRequest,
     JSONRPCResponse,
@@ -50,6 +49,10 @@
 
 logger = logging.getLogger(__name__)
 
+_ModelT = TypeVar("_ModelT", bound=BaseModel)
+
+_OK_STATUS = 200
+
 
 @dataclass
 class _SingleExchangeDispatchContext:
@@ -75,100 +78,65 @@ async def send_raw_request(
         raise NoBackChannelError(method)
 
     async def notify(self, method: str, params: Mapping[str, Any] | None) -> None:
+        # TODO(D-005a): buffer and stream as SSE once the JSON-vs-SSE response mode lands.
         return None
 
     async def progress(self, progress: float, total: float | None = None, message: str | None = None) -> None:
-        # TODO: no progressToken plumbing yet.
+        # TODO(D-005a): no progressToken plumbing yet; ships with the SSE response mode.
         return None
 
 
-class SingleExchangeDispatcher:
-    """Dispatcher for exactly one inbound JSON-RPC request over a single HTTP POST.
+def _typed(model: type[_ModelT], raw: Any) -> _ModelT | None:
+    """Validate the classifier's raw envelope value into a typed model.
 
-    The exception->wire boundary lives here (mirrors `JSONRPCDispatcher`'s
-    role). Implements the `Dispatcher` Protocol so `ServerRunner` /
-    `Connection` / `ServerSession` accept it; `run()` is never driven.
+    The classifier checks presence only; a value that fails shape validation
+    is treated as not supplied so the request still routes.
     """
-
-    def __init__(self, request: Request) -> None:
-        self._request = request
-        self._tctx = TransportContext(
-            kind="streamable-http",
-            can_send_request=False,
-            headers=request.headers,
-        )
-
-    async def send_raw_request(
-        self,
-        method: str,
-        params: Mapping[str, Any] | None,
-        opts: CallOptions | None = None,
-        *,
-        _related_request_id: RequestId | None = None,
-    ) -> dict[str, Any]:
-        raise NoBackChannelError(method)
-
-    async def notify(
-        self,
-        method: str,
-        params: Mapping[str, Any] | None,
-        *,
-        _related_request_id: RequestId | None = None,
-    ) -> None:
-        # TODO: buffer and stream as SSE once the response-mode design lands.
+    if raw is None:
+        return None
+    try:
+        return model.model_validate(raw, by_name=False)
+    except ValidationError:
         return None
 
-    async def run(
-        self,
-        on_request: OnRequest,
-        on_notify: OnNotify,
-        *,
-        task_status: anyio.abc.TaskStatus[None] = anyio.TASK_STATUS_IGNORED,
-    ) -> None:
-        raise RuntimeError("SingleExchangeDispatcher.run() is never driven; use handle()")
-
-    async def handle(self, req: JSONRPCRequest, on_request: OnRequest) -> JSONRPCResponse | JSONRPCError:
-        """Dispatch one request and map any exception to a `JSONRPCError`."""
-        dctx = _SingleExchangeDispatchContext(
-            transport=self._tctx,
-            request_id=req.id,
-            message_metadata=ServerMessageMetadata(request_context=self._request),
-        )
-        try:
-            result = await on_request(dctx, req.method, req.params)
-            return JSONRPCResponse(jsonrpc="2.0", id=req.id, result=result)
-        except MCPError as e:
-            return JSONRPCError(jsonrpc="2.0", id=req.id, error=e.error)
-        except ValidationError:
-            return JSONRPCError(
-                jsonrpc="2.0",
-                id=req.id,
-                error=ErrorData(code=INVALID_PARAMS, message="Invalid request parameters", data=""),
-            )
-        # TODO: consolidate the three exception->ErrorData copies once the
-        # code=0 compat pin in JSONRPCDispatcher is lifted.
-        except Exception:
-            logger.exception("handler for %r raised", req.method)
-            return JSONRPCError(
-                jsonrpc="2.0",
-                id=req.id,
-                error=ErrorData(code=INTERNAL_ERROR, message="Internal server error"),
-            )
+
+async def _write(
+    msg: JSONRPCResponse | JSONRPCError,
+    scope: Scope,
+    receive: Receive,
+    send: Send,
+    *,
+    extra_headers: Mapping[str, str] | None = None,
+) -> None:
+    """Serialise a JSON-RPC reply with the table-mapped HTTP status."""
+    status = ERROR_CODE_HTTP_STATUS.get(msg.error.code, _OK_STATUS) if isinstance(msg, JSONRPCError) else _OK_STATUS
+    body = msg.model_dump(mode="json", by_alias=True, exclude_none=True)
+    if isinstance(msg, JSONRPCError) and msg.id is None:
+        # JSON-RPC requires `id: null` to appear on the wire when the request
+        # id couldn't be parsed; `exclude_none` would otherwise drop it.
+        body["id"] = None
+    await Response(
+        json.dumps(body, separators=(",", ":")),
+        status_code=status,
+        media_type="application/json",
+        headers=dict(extra_headers) if extra_headers else None,
+    )(scope, receive, send)
 
 
 async def handle_modern_request(
     app: Server[Any],
     security_settings: TransportSecuritySettings | None,
-    protocol_version: str,
+    lifespan_state: Any,
     scope: Scope,
     receive: Receive,
     send: Send,
 ) -> None:
     """ASGI handler for a single stateless-era POST.
 
     Called from `StreamableHTTPSessionManager.handle_request` when the
-    `MCP-Protocol-Version` header is in `MODERN_PROTOCOL_VERSIONS`; the header
-    value is passed as `protocol_version`. Never sets `Mcp-Session-Id`.
+    `MCP-Protocol-Version` header names a modern revision; the manager enters
+    `app.lifespan` once at startup and passes the state in. Never sets
+    `Mcp-Session-Id`.
     """
     request = Request(scope, receive)
 
@@ -178,54 +146,42 @@ async def handle_modern_request(
         await err(scope, receive, send)
         return
 
-    # TODO: validate Accept header once the JSON-vs-SSE response-mode design is settled.
+    # TODO(D-005a): validate Accept once the JSON-vs-SSE response mode is settled.
 
     if request.method != "POST":
-        # TODO: GET/DELETE rejection (405 + -32601) lands with the validation ladder.
-        await Response(status_code=405, headers={"Allow": "POST"})(scope, receive, send)
+        rej = JSONRPCError(
+            jsonrpc="2.0",
+            id=None,
+            error=ErrorData(code=METHOD_NOT_FOUND, message=f"HTTP {request.method} not supported on this endpoint"),
+        )
+        await _write(rej, scope, receive, send, extra_headers={"Allow": "POST"})
         return
 
     body = await request.body()
     try:
         req = JSONRPCRequest.model_validate_json(body)
     except ValidationError:
-        msg = JSONRPCError(jsonrpc="2.0", id=None, error=ErrorData(code=PARSE_ERROR, message="Parse error"))
-        await Response(
-            msg.model_dump_json(by_alias=True),
-            status_code=400,
-            media_type="application/json",
-        )(scope, receive, send)
+        rej = JSONRPCError(jsonrpc="2.0", id=None, error=ErrorData(code=PARSE_ERROR, message="Parse error"))
+        await _write(rej, scope, receive, send)
         return
 
-    dispatcher = SingleExchangeDispatcher(request)
-    # TODO: per-request lifespan re-entry matches stateless_http=True today; revisit in #2893.
-    async with app.lifespan(app) as lifespan_state:
-        runner = ServerRunner(
-            server=app,
-            dispatcher=dispatcher,
-            lifespan_state=lifespan_state,
-            has_standalone_channel=False,
-            stateless=True,
-            dispatch_middleware=[otel_middleware],
+    verdict = classify_inbound_request({"method": req.method, "params": req.params}, headers=dict(request.headers))
+    if isinstance(verdict, InboundLadderRejection):
+        rej = JSONRPCError(
+            jsonrpc="2.0", id=req.id, error=ErrorData(code=verdict.code, message=verdict.message, data=verdict.data)
         )
-        runner.connection.protocol_version = protocol_version
-        try:
-            msg = await dispatcher.handle(req, runner._compose_on_request())  # type: ignore[reportPrivateUsage]
-        finally:
-            with anyio.move_on_after(_EXIT_STACK_CLOSE_TIMEOUT, shield=True) as cancel_scope:
-                try:
-                    await runner.connection.exit_stack.aclose()
-                except Exception:
-                    logger.exception("connection exit_stack cleanup raised")
-            if cancel_scope.cancelled_caught:
-                logger.warning(
-                    "connection exit_stack cleanup exceeded %s seconds; abandoning remaining callbacks",
-                    _EXIT_STACK_CLOSE_TIMEOUT,
-                )
-
-        # TODO: error.code -> HTTP status mapping is a follow-up; 200 for all JSONRPCError bodies for now.
-        await Response(
-            msg.model_dump_json(by_alias=True, exclude_none=True),
-            status_code=200,
-            media_type="application/json",
-        )(scope, receive, send)
+        await _write(rej, scope, receive, send)
+        return
+
+    connection = Connection.from_envelope(
+        verdict.protocol_version,
+        _typed(Implementation, verdict.client_info),
+        _typed(ClientCapabilities, verdict.client_capabilities),
+    )
+    dctx = _SingleExchangeDispatchContext(
+        transport=TransportContext(kind="streamable-http", can_send_request=False, headers=request.headers),
+        request_id=req.id,
+        message_metadata=ServerMessageMetadata(request_context=request),
+    )
+    msg = await serve_one(app, req, connection=connection, dctx=dctx, lifespan_state=lifespan_state)
+    await _write(msg, scope, receive, send)

src/mcp/server/lowlevel/server.py

@@ -56,9 +56,10 @@ async def main():
 from mcp.server.auth.provider import OAuthAuthorizationServerProvider, TokenVerifier
 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.connection import Connection
 from mcp.server.context import HandlerResult, ServerMiddleware, ServerRequestContext
 from mcp.server.models import InitializationOptions
-from mcp.server.runner import ServerRunner, otel_middleware
+from mcp.server.runner import serve_connection
 from mcp.server.streamable_http import EventStore
 from mcp.server.streamable_http_manager import StreamableHTTPASGIApp, StreamableHTTPSessionManager
 from mcp.server.transport_security import TransportSecuritySettings
@@ -438,12 +439,13 @@ async def run(
         # but also make tracing exceptions much easier during testing and when using
         # in-process servers.
         raise_exceptions: bool = False,
-        # When True, the server is stateless and
-        # clients can perform initialization with any node. The client must still follow
-        # the initialization lifecycle, but can do so with any available node
-        # rather than requiring initialization for each connection.
-        stateless: bool = False,
     ) -> None:
+        """Serve a single connection over the given streams until the read side closes.
+
+        Thin wrapper over `serve_connection` (L28): enters the server lifespan,
+        builds a `JSONRPCDispatcher` over the streams and a loop-mode
+        `Connection` on it, then drives the dispatcher to completion.
+        """
         async with self.lifespan(self) as lifespan_context:
             dispatcher: JSONRPCDispatcher[TransportContext] = JSONRPCDispatcher(
                 read_stream,
@@ -454,20 +456,14 @@ async def run(
                 # the initialized state instead of failing the init-gate.
                 inline_methods=frozenset({"initialize"}),
             )
-            runner = ServerRunner(
-                server=self,
-                dispatcher=dispatcher,
+            connection = Connection.for_loop(dispatcher)
+            await serve_connection(
+                self,
+                dispatcher,
+                connection=connection,
                 lifespan_state=lifespan_context,
                 init_options=initialization_options,
-                # Stateless HTTP has no standalone GET stream, so server-initiated
-                # requests on `runner.connection` must fail fast with
-                # `NoBackChannelError` rather than write to a channel that will
-                # never deliver a response.
-                has_standalone_channel=not stateless,
-                stateless=stateless,
-                dispatch_middleware=[otel_middleware],
             )
-            await runner.run()
 
     def streamable_http_app(
         self,