Address by-construction review: ServerSession reads standalone channel from Connection; hygiene

- ServerSession: drop standalone_outbound kwarg; read connection.outbound at the
  channel-select sites (one source of truth for the standalone channel; the
  documented two-arg form is now correct rather than mis-wiring)
- serve_loop: add to __all__; list in module docstring; soften the
  recipe-ownership claim
- _write: drop dead extra_headers kwarg
- _SingleExchangeDispatchContext.can_send_request: field(init=False) so True is
  unconstructible
- inbound: use UnsupportedProtocolVersionErrorData typed model for -32022 data;
  document why INTERNAL_ERROR is unmapped; anchor the header-validation TODO
- Sharpen the notification-POST rejection comment to cite the spec's
  cannot-accept branch
- Docstring corrections: session.py module docstring (per-request, not
  per-connection); Server.run() drops the internal tracking ref
- TODO anchors aligned to current tracking entries
- Tests: use public client_params property; complete ERROR_CODE_HTTP_STATUS
  parametrize; assert .error.code over message-text regex

Author: maxisbey

src/mcp/server/_streamable_http_modern.py

@@ -67,7 +67,7 @@ class _SingleExchangeDispatchContext:
     request_id: RequestId
     message_metadata: MessageMetadata
     cancel_requested: anyio.Event = field(default_factory=anyio.Event)
-    can_send_request: bool = False
+    can_send_request: bool = field(default=False, init=False)
 
     async def send_raw_request(
         self,
@@ -104,8 +104,6 @@ async def _write(
     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
@@ -118,7 +116,6 @@ async def _write(
         json.dumps(body, separators=(",", ":")),
         status_code=status,
         media_type="application/json",
-        headers=dict(extra_headers) if extra_headers else None,
     )(scope, receive, send)
 
 
@@ -163,10 +160,13 @@ async def handle_modern_request(
     try:
         req = JSONRPCRequest.model_validate(decoded)
     except ValidationError:
-        # Well-formed JSON that isn't a single request object (a notification,
-        # a response, a batch). TODO(L57): the 2026 HTTP path carries no
-        # client→server notifications (cancellation is via SSE close), so a
-        # plain rejection is currently sufficient.
+        # Well-formed JSON that isn't a single request object. The transport
+        # spec permits notification POSTs and gives the server two responses
+        # (202 accept / 4xx cannot-accept; streamable-http §Sending Messages
+        # item 5). The core protocol defines no client→server notifications
+        # over HTTP at 2026-07-28 (cancellation is SSE-stream close), so this
+        # entry takes the cannot-accept branch. TODO(L57): S4 owns the
+        # strict-vs-lenient choice.
         rej = JSONRPCError(
             jsonrpc="2.0",
             id=None,

src/mcp/server/lowlevel/server.py

@@ -226,7 +226,7 @@ def __init__(
         # Context-tier middleware: wraps every inbound request (including
         # `initialize`, lookup, validation, handler) with
         # `(ctx, method, params, call_next)`. Applied in `ServerRunner._on_request`.
-        # TODO(maxisbey): provisional - signature and semantics change with the
+        # TODO(L54): provisional - signature and semantics change with the
         # Context/middleware rework (covariant `Context[L]`, outbound seam) before
         # v2 final.
         self.middleware: list[ServerMiddleware[LifespanResultT]] = []
@@ -439,7 +439,7 @@ async def run(
     ) -> None:
         """Serve a single connection over the given streams until the read side closes.
 
-        Thin wrapper over `serve_loop` (L28): enters the server lifespan,
+        Thin wrapper over `serve_loop`: enters the server lifespan,
         then drives the loop. Transports with their own lifespan owner
         (the streamable-HTTP manager) call `serve_loop` directly instead.
         """

src/mcp/server/runner.py

@@ -5,8 +5,8 @@
 pure kernel: it holds a pre-populated `Connection` and reads
 `connection.protocol_version` / `connection.outbound` as facts. Driving a
 dispatcher loop and tearing down the connection live in the free-function
-drivers (`serve_connection`, `serve_one`); the entry constructs the
-`Connection`, the driver tears it down.
+drivers (`serve_connection`, `serve_loop`, `serve_one`); the entry constructs
+the `Connection`, the driver tears it down.
 
 `ServerRunner` holds a `Server` directly - `Server` is the registry.
 """
@@ -65,6 +65,7 @@
     "aclose_shielded",
     "otel_middleware",
     "serve_connection",
+    "serve_loop",
     "serve_one",
     "to_jsonrpc_response",
 ]
@@ -362,7 +363,7 @@ def _compose_server_middleware(
     def _make_context(
         self, dctx: DispatchContext[TransportContext], meta: RequestParamsMeta | None, protocol_version: str
     ) -> ServerRequestContext[LifespanT, Any]:
-        # TODO(maxisbey): remove for Context rework. Reads the SHTTP per-request
+        # TODO(L54): remove for Context rework. Reads the SHTTP per-request
         # data off the raw `dctx.message_metadata` carrier; replace with the
         # per-transport context once that lands.
         md = dctx.message_metadata
@@ -373,9 +374,9 @@ def _make_context(
         else:
             request = close_sse_stream = close_standalone_sse_stream = None
         # Per-request session: `dctx` is the request-scoped channel (auto-threads
-        # its own request_id on streamable HTTP); `connection.outbound` is the
-        # standalone channel. `related_request_id` on the public API selects.
-        session = ServerSession(dctx, self.connection, standalone_outbound=self.connection.outbound)
+        # its own request_id on streamable HTTP); the standalone channel is read
+        # off `connection.outbound`. `related_request_id` on the public API selects.
+        session = ServerSession(dctx, self.connection)
         return ServerRequestContext(
             session=session,
             lifespan_context=self.lifespan_state,
@@ -450,8 +451,8 @@ async def serve_loop(
     """Drive ``server`` in loop mode over a stream pair until the channel closes.
 
     Builds the loop-mode `JSONRPCDispatcher` + `Connection` and hands them to
-    `serve_connection`, so the dispatcher-construction recipe (notably the
-    `inline_methods={"initialize"}` rule) lives in one place. Callers that own
+    `serve_connection`, so loop-mode callers share one dispatcher-construction
+    recipe (notably the `inline_methods={"initialize"}` rule). Callers that own
     a lifespan (the streamable-HTTP manager) pass it in; callers that don't
     (`Server.run` for stdio/memory) enter the lifespan and then call this.
     """

src/mcp/server/session.py

@@ -1,12 +1,9 @@
 """`ServerSession`: server-to-client requests and notifications.
 
-A thin proxy over `JSONRPCDispatcher` and `Connection`. One instance per
-client connection (built by `ServerRunner`). Handlers reach it as
-`ctx.session` and use the typed helpers (`create_message`, `elicit_form`,
+A per-request proxy built by the kernel for each inbound request. Exposes the
+request-scoped outbound channel and the connection's standalone channel.
+Handlers reach it as `ctx.session` and use the typed helpers (`elicit_form`,
 `send_log_message`, ...) to call back to the client.
-
-The receive-loop, initialize handling, and per-request task isolation that
-used to live here are now owned by `JSONRPCDispatcher` and `ServerRunner`.
 """
 
 from typing import Any, TypeVar, overload
@@ -39,15 +36,8 @@ class ServerSession:
     never crosses the `Outbound` Protocol.
     """
 
-    def __init__(
-        self,
-        request_outbound: Outbound,
-        connection: Connection,
-        *,
-        standalone_outbound: Outbound | None = None,
-    ) -> None:
+    def __init__(self, request_outbound: Outbound, connection: Connection) -> None:
         self._request_outbound = request_outbound
-        self._standalone_outbound = standalone_outbound if standalone_outbound is not None else request_outbound
         self._connection = connection
 
     @property
@@ -81,7 +71,7 @@ async def send_request(
             pydantic.ValidationError: The peer's result does not match `result_type`.
         """
         related = metadata.related_request_id if metadata is not None else None
-        channel = self._request_outbound if related is not None else self._standalone_outbound
+        channel = self._request_outbound if related is not None else self._connection.outbound
         data = request.model_dump(by_alias=True, mode="json", exclude_none=True)
         opts: CallOptions = {}
         if request_read_timeout_seconds is not None:
@@ -101,7 +91,7 @@ async def send_notification(
         related_request_id: types.RequestId | None = None,
     ) -> None:
         """Send a typed server-to-client notification."""
-        channel = self._request_outbound if related_request_id is not None else self._standalone_outbound
+        channel = self._request_outbound if related_request_id is not None else self._connection.outbound
         data = notification.model_dump(by_alias=True, mode="json", exclude_none=True)
         await channel.notify(data["method"], data.get("params"))
 

src/mcp/server/streamable_http_manager.py

@@ -346,7 +346,7 @@ async def run_server(*, task_status: TaskStatus[None] = anyio.TASK_STATUS_IGNORE
                 await http_transport.handle_request(scope, receive, send)
         else:
             # Unknown or expired session ID - return 404 per MCP spec
-            # TODO: Align error code once spec clarifies
+            # TODO(L62): Align error code once spec clarifies
             # See: https://github.com/modelcontextprotocol/python-sdk/issues/1821
             logger.info(f"Rejected request with unknown or expired session ID: {request_mcp_session_id[:64]}")
             body = JSONRPCError(

src/mcp/shared/inbound.py

@@ -18,6 +18,7 @@
     CLIENT_CAPABILITIES_META_KEY,
     CLIENT_INFO_META_KEY,
     PROTOCOL_VERSION_META_KEY,
+    UnsupportedProtocolVersionErrorData,
 )
 from mcp.types.jsonrpc import (
     HEADER_MISMATCH,
@@ -40,6 +41,8 @@
 MCP_PROTOCOL_VERSION_HEADER: Final = "mcp-protocol-version"
 """Canonical lowercase name of the HTTP header carrying the MCP protocol version."""
 
+# INTERNAL_ERROR is deliberately unmapped (→ HTTP 200): the spec assigns no status to
+# -32603, and whether handler-origin errors get 5xx is an open S4 question — see TODO(L66).
 ERROR_CODE_HTTP_STATUS: Final[Mapping[int, int]] = MappingProxyType(
     {
         PARSE_ERROR: 400,
@@ -127,6 +130,7 @@ def classify_inbound_request(
             "client-capabilities envelope keys",
         )
 
+    # TODO(L59): also validate Mcp-Method / Mcp-Name per SEP-2243 §Server Validation
     if headers is not None and headers.get(MCP_PROTOCOL_VERSION_HEADER) != protocol_version:
         return InboundLadderRejection(
             code=HEADER_MISMATCH,
@@ -137,7 +141,9 @@ def classify_inbound_request(
         return InboundLadderRejection(
             code=UNSUPPORTED_PROTOCOL_VERSION,
             message="Unsupported protocol version",
-            data={"supported": list(supported_modern_versions), "requested": protocol_version},
+            data=UnsupportedProtocolVersionErrorData(
+                supported=list(supported_modern_versions), requested=protocol_version
+            ).model_dump(mode="json"),
         )
 
     return InboundModernRoute(

src/mcp/shared/jsonrpc_dispatcher.py

@@ -684,7 +684,7 @@ async def _handle_request(
                 # anyio absorbs the scope's own cancel at __exit__, and
                 # `cancelled_caught` (unlike `cancel_called`) guarantees the
                 # result write above did not happen - no double response.
-                # TODO(maxisbey): spec says SHOULD NOT respond after cancel;
+                # TODO(L38): spec says SHOULD NOT respond after cancel;
                 # the existing server always has, so match that for now.
                 answer_write_started = True
                 await self._write_error(req.id, ErrorData(code=0, message="Request cancelled"))
@@ -707,7 +707,7 @@ async def _handle_request(
                 await self._write_error(req.id, error)
             else:
                 logger.exception("handler for %r raised", req.method)
-                # TODO(L19): code=0 pins existing-server compat; JSON-RPC says
+                # TODO(L58): code=0 pins existing-server compat; JSON-RPC says
                 # INTERNAL_ERROR. Revisit per the suite's divergence entry.
                 await self._write_error(req.id, ErrorData(code=0, message=str(e)))
                 if self._raise_handler_exceptions:

tests/server/test_session.py

@@ -57,13 +57,13 @@ def _make_session(
     """Single-channel session: the stub is both request and standalone outbound."""
     client_info = Implementation(name="c", version="0") if capabilities is not None else None
     conn = Connection.from_envelope(protocol_version, client_info, capabilities, outbound=outbound)
-    return ServerSession(outbound, conn, standalone_outbound=outbound)
+    return ServerSession(outbound, conn)
 
 
 def _two_channel_session(request_ch: Outbound, standalone_ch: Outbound) -> ServerSession:
     """Distinct request/standalone outbounds so routing assertions can tell the channels apart."""
     conn = Connection.from_envelope(LATEST_PROTOCOL_VERSION, None, None, outbound=standalone_ch)
-    return ServerSession(request_ch, conn, standalone_outbound=standalone_ch)
+    return ServerSession(request_ch, conn)
 
 
 @pytest.mark.anyio
@@ -145,23 +145,6 @@ async def test_send_notification_routes_by_related_request_id():
     assert [m for m, _ in request_ch.notifications] == ["notifications/progress"]
 
 
-@pytest.mark.anyio
-async def test_standalone_outbound_defaults_to_request_outbound():
-    """SDK-defined: omitting `standalone_outbound` collapses both channels onto the
-    request one — the duplex case (stdio) where a single stream serves both.
-
-    `conn.outbound` is the no-channel sentinel here, so traffic landing on `only`
-    proves the default is the request outbound (the alternate impl — defaulting to
-    `conn.outbound` — would raise `NoBackChannelError`)."""
-    only = StubOutbound(result={"roots": []})
-    conn = Connection.from_envelope(LATEST_PROTOCOL_VERSION, None, None)
-    session = ServerSession(only, conn)
-    await session.send_request(types.ListRootsRequest(), types.ListRootsResult)
-    await session.send_notification(types.ToolListChangedNotification())
-    assert only.requests[0][0] == "roots/list"
-    assert only.notifications[0][0] == "notifications/tools/list_changed"
-
-
 @pytest.mark.anyio
 async def test_send_request_validates_the_client_result_against_the_surface_schema():
     """A spec-method result that fails the per-version surface schema raises
@@ -228,6 +211,6 @@ def test_protocol_version_proxies_connection():
     """SDK-defined: `session.protocol_version` reads through to the held `Connection`."""
     _ARBITRARY_VERSION = "sentinel-version"  # identity-only: any string the connection holds
     conn = Connection.from_envelope(_ARBITRARY_VERSION, None, None)
-    session = ServerSession(StubOutbound(), conn, standalone_outbound=conn.outbound)
+    session = ServerSession(StubOutbound(), conn)
     assert session.protocol_version == _ARBITRARY_VERSION
     assert session.client_params is None

tests/server/test_stateless_mode.py

@@ -47,7 +47,7 @@ def _no_channel_session(request_ch: StubOutbound | None = None) -> tuple[ServerS
     conn = Connection.from_envelope(LATEST_PROTOCOL_VERSION, None, None)
     assert conn.has_standalone_channel is False
     request = request_ch if request_ch is not None else StubOutbound()
-    return ServerSession(request, conn, standalone_outbound=conn.outbound), request
+    return ServerSession(request, conn), request
 
 
 @pytest.fixture
@@ -154,7 +154,7 @@ async def test_loop_connection_outbound_does_not_raise_no_back_channel():
     standalone = StubOutbound(result={"roots": []})
     conn = Connection.for_loop(standalone)
     assert conn.has_standalone_channel is True
-    session = ServerSession(StubOutbound(), conn, standalone_outbound=conn.outbound)
+    session = ServerSession(StubOutbound(), conn)
     result = await session.list_roots()  # pyright: ignore[reportDeprecated]
     assert isinstance(result, types.ListRootsResult)
     assert standalone.requests[0][0] == "roots/list"

tests/server/test_streamable_http_modern.py

@@ -129,7 +129,7 @@ async def test_handle_modern_request_routes_with_mis_shaped_envelope_client_info
     seen: list[object] = []
 
     async def greet(ctx: ServerRequestContext, params: PaginatedRequestParams) -> dict[str, Any]:
-        seen.append(ctx.session._connection.client_params)
+        seen.append(ctx.session.client_params)
         return {"ok": True}
 
     server: Server[Any] = Server("test")