Harden the dual-era stream loop's era-lock and rejection semantics (#3040)

Author: maxisbey

docs/advanced/subscriptions.md

@@ -46,6 +46,12 @@ Two more things the stream is *not*:
 * **It is not a replay log.** A dropped stream is gone; events published while nobody was connected are not queued. The client's contract is to re-listen and re-fetch what it cares about.
 * **It is not the 2025 path.** Clients on earlier protocol versions that called `resources/subscribe` are served by `ctx.session.send_resource_updated(uri)` — the `notify_*` methods reach `subscriptions/listen` streams only.
 
+!!! warning "Streamable HTTP only, for now"
+    `subscriptions/listen` is served on the streamable-HTTP transport. Over stdio (and other
+    stream-pair transports) a 2026-07-28 connection rejects it with METHOD_NOT_FOUND — the
+    open-stream semantics haven't been built for that transport yet, even though
+    `server/discover` still advertises the subscription capabilities there.
+
 ## One process is the default. More takes a bus
 
 Publishes travel from your handler to the open streams over a `SubscriptionBus`. The default is in-memory: one process, every stream in it. That is the right answer until you run replicas behind a load balancer — then a client's stream is pinned to one replica, and a publish on another replica has to reach it.

docs/client/protocol-versions.md

@@ -26,9 +26,9 @@ Either way you come out connected, and `client.protocol_version` tells you which
 That is the whole feature. One `Client`, any era of server, no branching in your code.
 
 !!! info
-    `MCPServer` answers `server/discover`, so against your own in-memory server `auto` always lands
-    on `2026-07-28`. The fallback only ever fires against a real pre-2026 server, which is exactly
-    when you want it to.
+    `MCPServer` answers `server/discover` on every transport — in-memory, stdio, streamable
+    HTTP — so against your own server `auto` always lands on `2026-07-28`. The fallback only
+    ever fires against a real pre-2026 server, which is exactly when you want it to.
 
 ## `mode="legacy"`
 

docs/migration.md

@@ -411,6 +411,8 @@ On the high-level `Client`, `client.server_capabilities`, `client.server_info`,
 
 In v1, connecting to a server always performed the `initialize` handshake. In v2, `Client` defaults to `mode='auto'`: on enter it probes `server/discover` and, if the server doesn't support it, falls back to the `initialize` handshake. Pass `mode='legacy'` to force the initialize handshake and reproduce v1's byte-identical pre-2026 behavior, or pass a modern protocol-version string (e.g. `mode='2026-07-28'`) to pin a version without probing.
 
+The probe is transport-independent: v2 servers answer it over stdio (and any other stream-pair transport) as well as streamable HTTP, so `mode='auto'` lands on `2026-07-28` against a v2 server on every transport. If your stdio workflow relies on server-initiated requests (sampling, push elicitation), pass `mode='legacy'` — a 2026-07-28 connection refuses them on every transport.
+
 For an in-process `Client(server)` (where `server` is a `Server` or `MCPServer` instance), `mode='auto'` dispatches calls directly through `DirectDispatcher` with no JSON-RPC framing. Pass `mode='legacy'` if you need the in-memory JSON-RPC transport that v1 used.
 
 `Client.send_ping()` is deprecated (ping is removed in 2026-07-28); pin `mode='legacy'` if you need it.

src/mcp/client/_probe.py

@@ -11,6 +11,12 @@
 the same path. Any non-``MCPError`` exception (network/connection errors,
 anyio cancellation, the ``RuntimeError`` from ``adopt()`` on no-mutual)
 propagates to the caller; an outage or in-process bug is never an era verdict.
+
+The fallback handshake itself can be answered with ``-32022`` — e.g. a probe
+that timed out client-side but succeeded on a slow-starting server locked the
+connection modern before the pipelined ``initialize`` arrived. That code is
+itself positive modern evidence (it names the server's versions), so it
+triggers one re-probe at a mutual version instead of failing the connect.
 """
 
 from __future__ import annotations
@@ -49,7 +55,8 @@ async def negotiate_auto(session: ClientSession) -> None:
 
     Raises:
         MCPError: The server is modern-only and shares no version with this
-            client (-32022 with a disjoint ``supported`` list).
+            client (-32022 with a disjoint ``supported`` list), or the
+            fallback handshake failed and one corrective re-probe did too.
         Exception: Any transport/network error from the probe propagates as-is.
     """
     version = LATEST_MODERN_VERSION
@@ -65,7 +72,22 @@ async def negotiate_auto(session: ClientSession) -> None:
                     continue
                 if supported is not None and not any(v in HANDSHAKE_PROTOCOL_VERSIONS for v in supported):
                     raise  # server is modern-only and disjoint — real incompatibility
-            await session.initialize()  # every other rpc-error → legacy (the denylist)
+            try:
+                await session.initialize()  # every other rpc-error → legacy (the denylist)
+            except MCPError as handshake_exc:
+                if handshake_exc.code != UNSUPPORTED_PROTOCOL_VERSION or attempt != 0:
+                    raise
+                # -32022 from the handshake is itself modern evidence: a probe
+                # that timed out client-side but succeeded on the server locked
+                # the connection modern before this initialize arrived. Re-probe
+                # once at a version the server names; the era is already
+                # settled, so the second probe answers without the slow start.
+                supported = _parse_supported(handshake_exc.error.data)
+                mutual = [v for v in MODERN_PROTOCOL_VERSIONS if v in (supported or ())]
+                if not mutual:
+                    raise
+                version = mutual[-1]
+                continue
             return
         # any other exception (httpx.TransportError, ConnectionError, anyio errors,
         # RuntimeError from adopt) → propagate

src/mcp/server/_streamable_http_modern.py

@@ -22,21 +22,18 @@
 import logging
 from collections.abc import Awaitable, Mapping
 from dataclasses import dataclass, field
-from typing import TYPE_CHECKING, Any, Final, TypeVar, cast
+from typing import TYPE_CHECKING, Any, Final, cast
 
 import anyio
 from anyio.streams.memory import MemoryObjectSendStream
 from mcp_types import (
     CLIENT_CAPABILITIES_META_KEY,
     CLIENT_INFO_META_KEY,
     HEADER_MISMATCH,
-    INTERNAL_ERROR,
     INVALID_REQUEST,
     PARSE_ERROR,
     PROTOCOL_VERSION_META_KEY,
-    ClientCapabilities,
     ErrorData,
-    Implementation,
     JSONRPCError,
     JSONRPCNotification,
     JSONRPCRequest,
@@ -45,13 +42,13 @@
     RequestId,
 )
 from mcp_types import methods as _methods
-from pydantic import BaseModel, ValidationError
+from pydantic import ValidationError
 from starlette.requests import Request
 from starlette.responses import Response
 from starlette.types import Receive, Scope, Send
 
 from mcp.server.connection import Connection
-from mcp.server.runner import serve_one
+from mcp.server.runner import modern_error_data, serve_one
 from mcp.server.streamable_http import check_accept_headers
 from mcp.server.transport_security import TransportSecurityMiddleware, TransportSecuritySettings
 from mcp.shared.dispatcher import CallOptions
@@ -65,7 +62,7 @@
     find_duplicated_routing_header,
     validate_mcp_param_headers,
 )
-from mcp.shared.jsonrpc_dispatcher import handler_exception_to_error_data, progress_token_from_params
+from mcp.shared.jsonrpc_dispatcher import progress_token_from_params
 from mcp.shared.message import MessageMetadata, ServerMessageMetadata
 from mcp.shared.transport_context import TransportContext
 
@@ -74,7 +71,6 @@
 
 logger = logging.getLogger(__name__)
 
-_ModelT = TypeVar("_ModelT", bound=BaseModel)
 
 _OK_STATUS = 200
 
@@ -125,37 +121,20 @@ async def progress(self, progress: float, total: float | None = None, message: s
         await self.notify("notifications/progress", params)
 
 
-def _typed(model: type[_ModelT], raw: Any) -> _ModelT | None:
-    """Validate the classifier's raw envelope value into a typed model.
-
-    Rung 1 guarantees the envelope key was present; a ``null`` or mis-shaped
-    value falls through to ``ValidationError`` and is treated as not supplied
-    so the request still routes.
-    """
-    try:
-        return model.model_validate(raw, by_name=False)
-    except ValidationError:
-        return None
-
-
 async def _to_jsonrpc_response(
     request_id: RequestId, coro: Awaitable[dict[str, Any]]
 ) -> JSONRPCResponse | JSONRPCError:
     """Await ``coro`` and wrap its outcome as the JSON-RPC reply for ``request_id``.
 
     The exception-to-wire boundary for the modern HTTP entry, composed around
-    `serve_one`. `MCPError` and `ValidationError` map via the shared
-    `handler_exception_to_error_data` ladder; any other exception is logged and
-    surfaced as `INTERNAL_ERROR` so handler internals never reach the wire.
+    `serve_one`: `modern_error_data` maps the shared ladder and surfaces
+    anything else as a generic `INTERNAL_ERROR` so handler internals never
+    reach the wire.
     """
     try:
         result = await coro
     except Exception as exc:
-        error = handler_exception_to_error_data(exc)
-        if error is None:
-            logger.exception("request handler raised")
-            error = ErrorData(code=INTERNAL_ERROR, message="Internal server error")
-        return JSONRPCError(jsonrpc="2.0", id=request_id, error=error)
+        return JSONRPCError(jsonrpc="2.0", id=request_id, error=modern_error_data(exc))
     return JSONRPCResponse(jsonrpc="2.0", id=request_id, result=result)
 
 
@@ -251,16 +230,16 @@ async def _tool_input_schema(
         logger.debug("Mcp-Param header validation skipped: the request envelope fails tools/list validation")
         return None
     seen_cursors: set[str] = set()
-    client_info = _typed(Implementation, verdict.client_info)
-    client_capabilities = _typed(ClientCapabilities, verdict.client_capabilities)
     dctx = _SingleExchangeDispatchContext(
         transport=TransportContext(kind="streamable-http", can_send_request=False, headers=request.headers),
         request_id=request_id,
         message_metadata=ServerMessageMetadata(request_context=request),
     )
     for _ in range(_MCP_PARAM_LIST_PAGE_CAP):
         # Fresh Connection per page: serve_one tears down the connection's exit stack on the way out.
-        connection = Connection.from_envelope(verdict.protocol_version, client_info, client_capabilities)
+        connection = Connection.from_envelope(
+            verdict.protocol_version, verdict.client_info, verdict.client_capabilities
+        )
         try:
             result = await serve_one(
                 app, dctx, "tools/list", list_params, connection=connection, lifespan_state=lifespan_state
@@ -409,8 +388,8 @@ async def handle_modern_request(
 
     connection = Connection.from_envelope(
         verdict.protocol_version,
-        _typed(Implementation, verdict.client_info),
-        _typed(ClientCapabilities, verdict.client_capabilities),
+        verdict.client_info,
+        verdict.client_capabilities,
     )
     dctx = _SingleExchangeDispatchContext(
         transport=TransportContext(kind="streamable-http", can_send_request=False, headers=request.headers),

src/mcp/server/connection.py

@@ -42,7 +42,7 @@
 )
 from mcp_types import methods as _methods
 from mcp_types.version import LATEST_HANDSHAKE_VERSION
-from pydantic import BaseModel
+from pydantic import BaseModel, ValidationError
 from typing_extensions import deprecated
 
 from mcp.shared.dispatcher import CallOptions, Outbound
@@ -68,6 +68,23 @@
 }
 
 
+_ModelT = TypeVar("_ModelT", bound=BaseModel)
+
+
+def _typed(model: type[_ModelT], raw: Any) -> _ModelT | None:
+    """Validate a raw envelope value into a typed model.
+
+    A missing, null or mis-shaped value falls through to `ValidationError`
+    and is treated as not supplied so the request still routes. Spec methods
+    are separately re-validated by the kernel's per-version params surface,
+    which types the reserved `_meta` keys strictly.
+    """
+    try:
+        return model.model_validate(raw, by_name=False)
+    except ValidationError:
+        return None
+
+
 def _notification_params(payload: dict[str, Any] | None, meta: Meta | None) -> dict[str, Any] | None:
     if not meta:
         return payload
@@ -100,26 +117,18 @@ async def notify(self, method: str, params: Mapping[str, Any] | None, opts: Call
 _NO_CHANNEL = _NoChannelOutbound()
 
 
-class NotifyOnlyOutbound:
+class NotifyOnlyOutbound(_NoChannelOutbound):
     """Connection-scoped `Outbound` that forwards notifications and refuses requests.
 
     Installed by `serve_dual_era_loop` for modern (2026-07-28+) connections
     over duplex stream transports: the pipe is real, so server notifications
     ride it, but the modern protocol forbids server-initiated JSON-RPC
-    requests, so `send_raw_request` refuses by construction.
+    requests, so `send_raw_request` (inherited) refuses by construction.
     """
 
     def __init__(self, outbound: Outbound) -> None:
         self._outbound = outbound
 
-    async def send_raw_request(
-        self,
-        method: str,
-        params: Mapping[str, Any] | None,
-        opts: CallOptions | None = None,
-    ) -> dict[str, Any]:
-        raise NoBackChannelError(method)
-
     async def notify(self, method: str, params: Mapping[str, Any] | None, opts: CallOptions | None = None) -> None:
         await self._outbound.notify(method, params, opts)
 
@@ -180,26 +189,34 @@ def __init__(
     def from_envelope(
         cls,
         protocol_version: str,
-        client_info: Implementation | None,
-        client_capabilities: ClientCapabilities | None,
+        client_info: Any,
+        client_capabilities: Any,
         *,
         outbound: Outbound = _NO_CHANNEL,
     ) -> Connection:
         """A born-ready connection populated from a request's `_meta` envelope.
 
-        `initialized` is set and the envelope's client info/capabilities (when
-        both supplied) are recorded as `client_params` so capability checks
-        work. `outbound` defaults to the no-channel sentinel for the
-        single-exchange HTTP path; duplex modern transports (e.g. stdio) pass
-        a notify-only wrapper around the dispatcher so server notifications
-        ride the pipe while server-initiated requests stay refused.
+        `protocol_version` must be an already-validated version string - the
+        inbound classification ladder owns rejecting non-string or unsupported
+        values. `client_info` and `client_capabilities` are the raw envelope
+        values: this constructor owns turning them into connection identity,
+        identically on every modern entry, so a mis-shaped value degrades to
+        not-supplied rather than failing the request. `initialized`
+        is set and the info/capabilities (when both supplied and well-formed)
+        are recorded as `client_params` so capability checks work. `outbound`
+        defaults to the no-channel sentinel for the single-exchange HTTP path;
+        duplex modern transports (e.g. stdio) pass a notify-only wrapper
+        around the dispatcher so server notifications ride the pipe while
+        server-initiated requests stay refused.
         """
+        info = _typed(Implementation, client_info)
+        capabilities = _typed(ClientCapabilities, client_capabilities)
         client_params = None
-        if client_info is not None and client_capabilities is not None:
+        if info is not None and capabilities is not None:
             client_params = InitializeRequestParams(
                 protocol_version=protocol_version,
-                capabilities=client_capabilities,
-                client_info=client_info,
+                capabilities=capabilities,
+                client_info=info,
             )
         connection = cls(outbound, protocol_version=protocol_version, client_params=client_params)
         connection.initialized.set()
@@ -230,7 +247,12 @@ def for_loop(
     def has_standalone_channel(self) -> bool:
         """Whether this connection has a real back-channel for server-initiated
         messages. Derived from `outbound` - the no-channel sentinel is the only
-        case that doesn't."""
+        case that doesn't.
+
+        Channel presence, not request permission: a modern (2026-07-28+)
+        duplex connection has a channel that carries notifications while
+        `send_raw_request` still refuses, because the protocol forbids
+        server-initiated requests."""
         return self.outbound is not _NO_CHANNEL
 
     @property
@@ -255,7 +277,9 @@ async def send_raw_request(
 
         Raises:
             MCPError: The peer responded with an error.
-            NoBackChannelError: `has_standalone_channel` is `False`.
+            NoBackChannelError: no back-channel for server-initiated requests -
+                `has_standalone_channel` is `False`, or a modern (2026-07-28+)
+                connection, where the protocol forbids them.
         """
         return await self.outbound.send_raw_request(method, params, opts)
 
@@ -316,7 +340,9 @@ async def ping(self, *, meta: Meta | None = None, opts: CallOptions | None = Non
 
         Raises:
             MCPError: The peer responded with an error.
-            NoBackChannelError: `has_standalone_channel` is `False`.
+            NoBackChannelError: no back-channel for server-initiated requests -
+                `has_standalone_channel` is `False`, or a modern (2026-07-28+)
+                connection, where the protocol forbids them.
         """
         await self.send_raw_request("ping", dump_params(None, meta), opts)
 

src/mcp/server/lowlevel/server.py

@@ -691,8 +691,8 @@ async def run(
 
         Thin wrapper over `serve_dual_era_loop`: enters the server lifespan,
         then drives the loop, serving the legacy handshake era and the modern
-        per-request-envelope era (the first era-distinctive message locks the
-        connection). Transports with their own lifespan owner (the
+        per-request-envelope era (the first era-distinctive message to succeed
+        locks the connection). Transports with their own lifespan owner (the
         streamable-HTTP manager) call `serve_loop` directly instead.
         """
         async with self.lifespan(self) as lifespan_context: