From fee8c52acc7aa937e5bc5017ca344130f3689cf4 Mon Sep 17 00:00:00 2001 From: Max Isbey <224885523+maxisbey@users.noreply.github.com> Date: Thu, 23 Jul 2026 14:18:30 +0000 Subject: [PATCH 1/4] Serve the 2026-07-28 protocol over stdio by deciding the era from the opening request The stdio driver derived a connection's protocol era from whichever request finished first, so it could never host a request that does not return: subscriptions/listen was hard-refused with -32601 even though the same connection advertised list-changed capabilities, and a legacy handshake arriving during a slow 2026 request was accepted and locked the connection out from under it. Decide the era from the client's first request instead, once, before any handler runs: a request carrying the 2026-07-28 per-request envelope opens a 2026 connection, anything else (the initialize handshake) opens a 2025 one, and the deciding frame is replayed into the chosen serving loop. A conflicting later claim is refused in one place (initialize on a 2026 connection gets -32022 naming the served versions; an enveloped request on a handshake connection gets -32600). The listen refusal is deleted; the handler was always transport-agnostic. --- docs/whats-new.md | 2 +- src/mcp/server/runner.py | 263 +++++++++++++++++++----------------- tests/server/test_runner.py | 178 ++++++++++-------------- 3 files changed, 214 insertions(+), 229 deletions(-) diff --git a/docs/whats-new.md b/docs/whats-new.md index 9c7a9be91..3f1188f8f 100644 --- a/docs/whats-new.md +++ b/docs/whats-new.md @@ -191,7 +191,7 @@ That file is the pitch in one place: one server, one `Resolve`-backed tool, and ### Change notifications become one stream -At 2026-07-28 the standalone HTTP GET stream and `resources/subscribe` are replaced by `subscriptions/listen`: the client opens one long-lived stream and names the notification kinds it wants. `MCPServer` serves it out of the box; you publish with `await ctx.notify_resource_updated(uri)` (and `notify_tools_changed()`, and so on), and multi-replica deployments plug in a shared `SubscriptionBus`. On the client (since `2.0.0b2`), `async with client.listen(...)` opens the stream: the filter goes in as keyword arguments, typed change events come back, and `sub.honored` is the subset the server agreed to deliver. One honest caveat: over stdio the server does not serve the stream yet. +At 2026-07-28 the standalone HTTP GET stream and `resources/subscribe` are replaced by `subscriptions/listen`: the client opens one long-lived stream and names the notification kinds it wants. `MCPServer` serves it out of the box; you publish with `await ctx.notify_resource_updated(uri)` (and `notify_tools_changed()`, and so on), and multi-replica deployments plug in a shared `SubscriptionBus`. On the client (since `2.0.0b2`), `async with client.listen(...)` opens the stream: the filter goes in as keyword arguments, typed change events come back, and `sub.honored` is the subset the server agreed to deliver. **[Subscriptions](handlers/subscriptions.md)** covers publishing and serving, **[its Clients twin](client/subscriptions.md)** the watching end, and **[Deploy & scale](run/deploy.md)** the bus. diff --git a/src/mcp/server/runner.py b/src/mcp/server/runner.py index 98bc133a7..5e80a1dcd 100644 --- a/src/mcp/server/runner.py +++ b/src/mcp/server/runner.py @@ -14,10 +14,11 @@ from __future__ import annotations import logging -from collections.abc import Awaitable, Mapping +from collections.abc import AsyncIterator, Awaitable, Mapping +from contextlib import asynccontextmanager from dataclasses import KW_ONLY, dataclass, replace from functools import cached_property, partial -from typing import TYPE_CHECKING, Any, Generic, Literal, cast +from typing import TYPE_CHECKING, Any, Generic, cast import anyio import anyio.abc @@ -37,6 +38,7 @@ Implementation, InitializeRequestParams, InitializeResult, + JSONRPCRequest, RequestId, RequestParams, RequestParamsMeta, @@ -605,91 +607,153 @@ async def serve_dual_era_loop( init_options: InitializationOptions | None = None, raise_exceptions: bool = False, ) -> None: - """Drive `server` over a duplex stream pair, serving both protocol eras. - - The stream-pair counterpart of the modern HTTP entry's era router. Era is - a property of the connection, decided by how the client opens it, and - mid-stream switching is undefined - so the first era-distinctive message - to SUCCEED locks the connection (matching the typescript-sdk): - - - A successful `initialize` locks legacy: the connection behaves exactly - like `serve_loop` for its lifetime, and modern envelope traffic is then - rejected with INVALID_REQUEST. `initialize` never routes modern - the - method is legacy-distinctive by definition - even when a confused - client stamps the envelope keys on it. - - A request whose `_meta` declares the modern protocol version - or - `server/discover`, a modern-only method - is classified - (`classify_inbound_request`) and served single-exchange via `serve_one` - with a born-ready per-request `Connection`, the same dispatch model as - the modern HTTP entry. The first such request to succeed locks the - connection modern; a later `initialize` is then rejected with - UNSUPPORTED_PROTOCOL_VERSION naming the modern versions. - - Modern connections push notifications over the duplex pipe but refuse - server-initiated requests on both channels (the modern protocol forbids - them). A request that fails - rejected classification, malformed envelope - content, unknown method - never locks either era, so a failed probe - leaves the legacy handshake available: released auto-negotiating clients - fall back on any error code except -32022, and that code is only emitted - for genuine version negotiation or for `initialize` on an - already-modern connection. - - The era lock rides the request's own dispatch. For the inline methods - (`initialize`, `server/discover`) that completes before the next frame is - read, so the canonical probe-then-go flow is race-free; a pinned-modern - client that pipelines frames ahead of its first response should expect - envelope-less notifications sent in that window to be dropped. The lock - settles exactly once: a request from the other era that was already in - flight when the lock committed may still complete and its response - stands, but the era does not move; and a success the peer cancelled away - (it sees "Request cancelled", not the result) does not lock either. + """Drive `server` over a duplex stream pair, in the era the client opens with. + + The client's first request decides the connection's protocol era, once: + a request carrying the 2026-07-28 per-request `_meta` envelope opens a + modern connection, and anything else - the `initialize` handshake, which + does not exist at 2026 versions even when a client stamps the envelope on + it - opens a legacy one. The deciding frame is replayed into the chosen + serving loop along with everything the client sent before it. A later + claim from the other era is refused: `initialize` on a modern connection + gets UNSUPPORTED_PROTOCOL_VERSION naming the served versions, and an + enveloped request on a legacy connection gets INVALID_REQUEST. """ + # This loop owns both streams from the moment it is called, so the write + # stream is closed even if the client leaves before sending any request. + try: + async with _replay_from_opening_request(read_stream) as (opening, replayed): + opens_modern = ( + opening is not None and opening.method != "initialize" and _has_modern_envelope(opening.params) + ) + if opens_modern: + await _serve_modern_stream( + server, replayed, write_stream, lifespan_state=lifespan_state, raise_exceptions=raise_exceptions + ) + else: + await _serve_legacy_stream( + server, + replayed, + write_stream, + lifespan_state=lifespan_state, + session_id=session_id, + init_options=init_options, + raise_exceptions=raise_exceptions, + ) + finally: + await write_stream.aclose() + + +@asynccontextmanager +async def _replay_from_opening_request( + read_stream: ReadStream[SessionMessage | Exception], +) -> AsyncIterator[tuple[JSONRPCRequest | None, ReadStream[SessionMessage | Exception]]]: + """Peek at the client's first request without consuming anything. + + Reads frames until the first JSON-RPC request arrives, then yields that + request together with a stream that replays every frame read so far and + relays the rest of `read_stream` behind it. The request is `None` if the + client closed the channel before sending any request. + """ + peeked: list[SessionMessage | Exception] = [] + opening: JSONRPCRequest | None = None + replay_send, replayed = anyio.create_memory_object_stream[SessionMessage | Exception]() + + async def replay_then_relay() -> None: + async with replay_send: + for item in peeked: + await replay_send.send(item) + async for item in read_stream: + await replay_send.send(item) + + # This helper takes ownership of `read_stream` from the serving loop, so + # every exit - including cancellation while awaiting the first request - + # closes it and the replay channel. + try: + async for item in read_stream: + peeked.append(item) + if isinstance(item, SessionMessage) and isinstance(item.message, JSONRPCRequest): + opening = item.message + break + async with anyio.create_task_group() as tg: + tg.start_soon(replay_then_relay) + yield opening, replayed + tg.cancel_scope.cancel() + finally: + await read_stream.aclose() + replay_send.close() + replayed.close() + + +async def _serve_legacy_stream( + server: Server[LifespanT], + read_stream: ReadStream[SessionMessage | Exception], + write_stream: WriteStream[SessionMessage], + *, + lifespan_state: LifespanT, + session_id: str | None, + init_options: InitializationOptions | None, + raise_exceptions: bool, +) -> None: + """Serve a 2025 handshake connection; enveloped requests are refused.""" dispatcher: JSONRPCDispatcher[TransportContext] = JSONRPCDispatcher( read_stream, write_stream, raise_handler_exceptions=raise_exceptions, - # `initialize` inline for the same pipelining reason as `serve_loop`; - # `server/discover` inline so the modern era lock commits before the - # next pipelined message is read. - inline_methods=frozenset({"initialize", "server/discover"}), + # `initialize` inline for the same pipelining reason as `serve_loop`. + inline_methods=frozenset({"initialize"}), + ) + connection = Connection.for_loop(dispatcher, session_id=session_id) + runner = ServerRunner(server, connection, lifespan_state, init_options=init_options) + + async def on_request( + dctx: DispatchContext[TransportContext], method: str, params: Mapping[str, Any] | None + ) -> dict[str, Any]: + if method != "initialize" and _has_modern_envelope(params): + raise MCPError( + code=INVALID_REQUEST, + message="connection was opened with the initialize handshake; " + "2026-07-28 envelope requests are not accepted on it", + ) + return await runner.on_request(dctx, method, params) + + try: + await dispatcher.run(on_request, runner.on_notify) + finally: + await aclose_shielded(connection) + + +async def _serve_modern_stream( + server: Server[LifespanT], + read_stream: ReadStream[SessionMessage | Exception], + write_stream: WriteStream[SessionMessage], + *, + lifespan_state: LifespanT, + raise_exceptions: bool, +) -> None: + """Serve a 2026-07-28 connection: every request carries its own envelope.""" + dispatcher: JSONRPCDispatcher[TransportContext] = JSONRPCDispatcher( + read_stream, write_stream, raise_handler_exceptions=raise_exceptions ) - loop_connection = Connection.for_loop(dispatcher, session_id=session_id) - loop_runner = ServerRunner(server, loop_connection, lifespan_state, init_options=init_options) - standalone_outbound = NotifyOnlyOutbound(dispatcher) - era: Literal["unlocked", "legacy", "modern"] = "unlocked" - modern_version = LATEST_MODERN_VERSION - - def era_settles(dctx: DispatchContext[TransportContext]) -> bool: - # The one definition of "this request may lock the era": it settled as - # a client-visible success on a still-unlocked connection. The lock is - # monotone - the first success wins, so a straggling request from the - # other era can never overwrite a committed lock. A pending peer - # cancel means the dispatcher is about to replace this response with - # "Request cancelled": the client never sees the success, no lock. - return era == "unlocked" and not dctx.cancel_requested.is_set() - - async def serve_modern( + outbound = NotifyOnlyOutbound(dispatcher) + + async def on_request( dctx: DispatchContext[TransportContext], method: str, params: Mapping[str, Any] | None ) -> dict[str, Any]: - nonlocal era, modern_version + if method == "initialize": + raise MCPError( + code=UNSUPPORTED_PROTOCOL_VERSION, + message="connection is serving the 2026-07-28 protocol; the initialize handshake is not accepted", + data=_initialize_after_modern_data(params), + ) route = classify_inbound_request({"method": method, "params": params}) if isinstance(route, InboundLadderRejection): raise MCPError(code=route.code, message=route.message, data=route.data) - if method == "subscriptions/listen": - # The registered listen handler assumes the HTTP entry's stream - # semantics; served over a stream pair it would wedge. Reject until - # this transport grows its own listen design. - raise MCPError( - code=METHOD_NOT_FOUND, message="subscriptions/listen is not served over this transport", data=method - ) connection = Connection.from_envelope( - route.protocol_version, - route.client_info, - route.client_capabilities, - outbound=standalone_outbound, + route.protocol_version, route.client_info, route.client_capabilities, outbound=outbound ) try: - result = await serve_one( + return await serve_one( server, _NoServerRequestsDispatchContext(dctx), method, @@ -698,68 +762,25 @@ async def serve_modern( lifespan_state=lifespan_state, ) except (MCPError, ValidationError): - # The dispatcher's shared ladder maps these to the same wire error - # the modern HTTP entry produces. + # The dispatcher's shared ladder maps these to the wire error. raise except Exception as exc: if raise_exceptions: raise error = modern_error_data(exc) raise MCPError(code=error.code, message=error.message, data=error.data) from exc - if era_settles(dctx): - era, modern_version = "modern", route.protocol_version - return result - - async def on_request( - dctx: DispatchContext[TransportContext], method: str, params: Mapping[str, Any] | None - ) -> dict[str, Any]: - nonlocal era - if era == "legacy": - if _has_modern_envelope(params): - raise MCPError( - code=INVALID_REQUEST, - message="connection is locked to the legacy handshake era; " - "modern envelope requests are not accepted", - ) - # Bare modern-only methods (e.g. `server/discover`) fall through to - # the loop runner's per-version surface validation - the same - # METHOD_NOT_FOUND a handshake-only server produced, byte for byte. - return await loop_runner.on_request(dctx, method, params) - if era == "modern": - if method == "initialize": - raise MCPError( - code=UNSUPPORTED_PROTOCOL_VERSION, - message="connection already negotiated a modern protocol version", - data=_initialize_after_modern_data(params), - ) - return await serve_modern(dctx, method, params) - # Unlocked. `initialize` is legacy-distinctive by definition (the - # method does not exist at modern versions), so it takes the handshake - # path even when the envelope keys are stamped on it. - if method != "initialize" and (method == "server/discover" or _has_modern_envelope(params)): - return await serve_modern(dctx, method, params) - result = await loop_runner.on_request(dctx, method, params) - if method == "initialize" and era_settles(dctx): - # Lock only on success: a failed handshake leaves both eras open. - era = "legacy" - return result async def on_notify(dctx: DispatchContext[TransportContext], method: str, params: Mapping[str, Any] | None) -> None: - if era != "modern": - return await loop_runner.on_notify(dctx, method, params) - # The envelope is request-only, so notifications inherit the - # connection's locked version. - connection = Connection.from_envelope(modern_version, None, None, outbound=standalone_outbound) + # The envelope is request-only, so a notification runs at the latest + # served version; the modern protocol has nothing version-specific here. + connection = Connection.from_envelope(LATEST_MODERN_VERSION, None, None, outbound=outbound) notify_runner = ServerRunner(server, connection, lifespan_state) try: await notify_runner.on_notify(_NoServerRequestsDispatchContext(dctx), method, params) finally: await aclose_shielded(connection) - try: - await dispatcher.run(on_request, on_notify) - finally: - await aclose_shielded(loop_connection) + await dispatcher.run(on_request, on_notify) async def serve_one( diff --git a/tests/server/test_runner.py b/tests/server/test_runner.py index 174cb853a..484593a3d 100644 --- a/tests/server/test_runner.py +++ b/tests/server/test_runner.py @@ -68,6 +68,7 @@ serve_one, ) from mcp.server.session import ServerSession +from mcp.server.subscriptions import SUBSCRIPTION_ID_META_KEY, InMemorySubscriptionBus, ListenHandler from mcp.shared.dispatcher import CallOptions from mcp.shared.exceptions import MCPError, NoBackChannelError from mcp.shared.jsonrpc_dispatcher import JSONRPCDispatcher @@ -1539,7 +1540,7 @@ async def test_dual_era_loop_initialize_locks_legacy_and_rejects_modern_traffic( assert result["tools"][0]["name"] == "t" assert discover_exc.value.error.code == INVALID_REQUEST assert envelope_exc.value.error.code == INVALID_REQUEST - assert "locked to the legacy handshake era" in discover_exc.value.error.message + assert "opened with the initialize handshake" in envelope_exc.value.error.message @pytest.mark.anyio @@ -1558,44 +1559,49 @@ async def test_dual_era_loop_bare_discover_after_legacy_lock_is_byte_identical(s @pytest.mark.anyio -async def test_dual_era_loop_unsupported_modern_version_rejects_without_locking(server: SrvT): - """A probe at an unknown modern version gets -32022 with the supported - list, and the rejection does not lock the era: the legacy handshake still - succeeds afterwards (the released auto clients' retry/fallback contract).""" +async def test_dual_era_loop_unsupported_modern_version_gets_the_supported_list(server: SrvT): + """A probe at an unknown modern version is answered -32022 with the + supported list; the connection is a 2026 one regardless, so the client + picks a listed version instead of falling back to `initialize` - which the + modern connection refuses with -32022 too, exactly as stdio.mdx directs.""" async with dual_era_client(server) as (client, _): with pytest.raises(MCPError) as exc_info: await client.send_raw_request("server/discover", _modern_params(version="2099-01-01")) - init = await client.send_raw_request("initialize", _initialize_params()) - assert init["protocolVersion"] == LATEST_HANDSHAKE_VERSION + with pytest.raises(MCPError) as init_exc: + await client.send_raw_request("initialize", _initialize_params()) assert exc_info.value.error.code == UNSUPPORTED_PROTOCOL_VERSION assert exc_info.value.error.data == { "supported": list(MODERN_PROTOCOL_VERSIONS), "requested": "2099-01-01", } + assert init_exc.value.error.code == UNSUPPORTED_PROTOCOL_VERSION @pytest.mark.anyio -async def test_dual_era_loop_bare_discover_rejects_without_locking(server: SrvT): - """A `server/discover` with no envelope triple is INVALID_PARAMS - never - -32022, so a released auto client's code-keyed fallback predicate takes the - legacy branch - and the connection can still complete the handshake.""" +async def test_dual_era_loop_bare_discover_opens_legacy_and_keeps_the_handshake_available(server: SrvT): + """`server/discover` without the envelope is not 2026 vocabulary, so it + opens a handshake connection: the answer is METHOD_NOT_FOUND (never -32022, + the one code a probing client must not treat as "fall back"), and the + client's fallback `initialize` then succeeds on the same connection.""" async with dual_era_client(server) as (client, _): with pytest.raises(MCPError) as exc_info: await client.send_raw_request("server/discover", None) init = await client.send_raw_request("initialize", _initialize_params()) assert init["protocolVersion"] == LATEST_HANDSHAKE_VERSION - assert exc_info.value.error.code == INVALID_PARAMS + assert exc_info.value.error.code == METHOD_NOT_FOUND assert exc_info.value.error.code != UNSUPPORTED_PROTOCOL_VERSION @pytest.mark.anyio -async def test_dual_era_loop_ping_before_any_lock_stays_exempt_and_neutral(server: SrvT): - """A pre-handshake `ping` is answered (the init-gate exemption) and does - not lock an era: the connection can still go modern.""" +async def test_dual_era_loop_an_envelopeless_first_request_opens_a_legacy_connection(server: SrvT): + """A first request without the 2026 envelope - here a pre-handshake `ping`, + answered under the init-gate exemption - is handshake-era vocabulary and + opens a legacy connection: enveloped requests are refused after it.""" async with dual_era_client(server) as (client, _): assert await client.send_raw_request("ping", None) == {} - result = await client.send_raw_request("tools/list", _modern_params()) - assert result["tools"][0]["name"] == "t" + with pytest.raises(MCPError) as exc_info: + await client.send_raw_request("tools/list", _modern_params()) + assert exc_info.value.error.code == INVALID_REQUEST @pytest.mark.anyio @@ -1610,34 +1616,43 @@ async def test_dual_era_loop_modern_request_without_envelope_rejects(server: Srv @pytest.mark.anyio -async def test_dual_era_loop_rejects_subscriptions_listen_on_modern(server: SrvT): - """`subscriptions/listen` is rejected before dispatch on the stream-pair - modern path (the registered handler assumes the HTTP entry's stream - semantics) - and like every failed request it does not lock the era, so - the legacy handshake stays available.""" - async with dual_era_client(server) as (client, _): - with pytest.raises(MCPError) as exc_info: - await client.send_raw_request("subscriptions/listen", _modern_params()) - init = await client.send_raw_request("initialize", _initialize_params()) - assert init["protocolVersion"] == LATEST_HANDSHAKE_VERSION - assert exc_info.value.error.code == METHOD_NOT_FOUND - assert "not served over this transport" in exc_info.value.error.message +async def test_dual_era_loop_serves_subscriptions_listen_over_the_stream_pair(): + """`subscriptions/listen` is served over the duplex stream like any other + modern request: the acknowledgement notification rides the pipe first, + and the server ending the stream yields the request's graceful-close + result, stamped with the subscription id.""" + listen = ListenHandler(InMemorySubscriptionBus()) + listener = Server(name="listener-server", version="0.0.1", on_subscriptions_listen=listen) + async with dual_era_client(listener) as (client, recorder): + result: dict[str, Any] = {} + + async def open_listen() -> None: + params = _modern_params(notifications={"toolsListChanged": True}) + result.update(await client.send_raw_request("subscriptions/listen", params)) + + async with anyio.create_task_group() as tg: + tg.start_soon(open_listen) + await recorder.notified.wait() + listen.close() + assert recorder.notifications[0][0] == "notifications/subscriptions/acknowledged" + assert result["resultType"] == "complete" + assert SUBSCRIPTION_ID_META_KEY in result["_meta"] @pytest.mark.anyio -async def test_dual_era_loop_malformed_envelope_content_never_locks(server: SrvT): - """The envelope triple with mis-shaped values fails the request but never - locks the era: the lock commits only when a modern request SUCCEEDS, so a - buggy client's initialize fallback still works (it must never see -32022 - for a request that failed).""" +async def test_dual_era_loop_malformed_envelope_content_is_a_modern_era_error(server: SrvT): + """An envelope with mis-shaped values still declares the 2026 era: the + request is rejected INVALID_PARAMS in modern vocabulary, and the connection + is a modern one, so a handshake sent afterwards is refused rather than + served on a connection that already speaks the other era.""" params: dict[str, Any] = {"_meta": {**_modern_envelope(), CLIENT_INFO_META_KEY: 42}} async with dual_era_client(server) as (client, _): with pytest.raises(MCPError) as exc_info: await client.send_raw_request("tools/list", params) - init = await client.send_raw_request("initialize", _initialize_params()) - assert init["protocolVersion"] == LATEST_HANDSHAKE_VERSION + with pytest.raises(MCPError) as init_exc: + await client.send_raw_request("initialize", _initialize_params()) assert exc_info.value.error.code == INVALID_PARAMS - assert exc_info.value.error.code != UNSUPPORTED_PROTOCOL_VERSION + assert init_exc.value.error.code == UNSUPPORTED_PROTOCOL_VERSION @pytest.mark.anyio @@ -1661,32 +1676,32 @@ async def test_dual_era_loop_pair_only_envelope_serves_modern_and_locks(server: @pytest.mark.anyio -async def test_dual_era_loop_version_without_capabilities_rejects_naming_the_key_and_never_locks(server: SrvT): +async def test_dual_era_loop_version_without_capabilities_rejects_naming_the_key(server: SrvT): """A `_meta` declaring the protocol version but missing the required client-capabilities key routes modern - never the legacy path with its generic 'Invalid request parameters' - and is rejected INVALID_PARAMS - naming the missing key; like every failed classification it locks no era, - so the legacy handshake stays available.""" + naming the missing key; a correctly enveloped request then serves.""" params = _modern_params() del params["_meta"][CLIENT_CAPABILITIES_META_KEY] async with dual_era_client(server) as (client, _): with pytest.raises(MCPError) as exc_info: await client.send_raw_request("tools/list", params) - init = await client.send_raw_request("initialize", _initialize_params()) - assert init["protocolVersion"] == LATEST_HANDSHAKE_VERSION + result = await client.send_raw_request("tools/list", _modern_params()) + assert result["tools"][0]["name"] == "t" assert exc_info.value.error.code == INVALID_PARAMS assert CLIENT_CAPABILITIES_META_KEY in exc_info.value.error.message @pytest.mark.anyio -async def test_dual_era_loop_failed_modern_request_never_locks(server: SrvT): - """A well-formed modern request for an unknown method fails without - locking; the next modern request locks on its own success.""" +async def test_dual_era_loop_failed_modern_request_leaves_the_connection_modern(server: SrvT): + """A well-formed modern request for an unknown method fails on its own; + the connection is a 2026 one either way, so the next modern request + serves.""" async with dual_era_client(server) as (client, _): with pytest.raises(MCPError) as exc_info: await client.send_raw_request("nope/missing", _modern_params()) - init = await client.send_raw_request("initialize", _initialize_params()) - assert init["protocolVersion"] == LATEST_HANDSHAKE_VERSION + result = await client.send_raw_request("tools/list", _modern_params()) + assert result["tools"][0]["name"] == "t" assert exc_info.value.error.code == METHOD_NOT_FOUND @@ -1774,11 +1789,11 @@ async def wants_roots(ctx: Ctx, params: RequestParams | None) -> dict[str, Any]: @pytest.mark.anyio -async def test_dual_era_loop_late_modern_success_does_not_overwrite_a_committed_legacy_lock(): - """The era settles exactly once, on the FIRST client-visible success: a - modern request that was already in flight when a legacy handshake - committed may still complete - its response stands - but the connection - stays legacy, so the handshaked client is never stranded.""" +async def test_dual_era_loop_initialize_during_an_in_flight_modern_request_is_refused(): + """The era is fixed by the opening request, not by what has completed: a + handshake arriving while the opening modern request is still parked in + its handler is refused with -32022, and the parked request finishes on + the modern connection it opened.""" entered = anyio.Event() release = anyio.Event() @@ -1796,60 +1811,12 @@ async def modern_call() -> None: async with anyio.create_task_group() as tg: tg.start_soon(modern_call) - # The modern dispatch is parked in its handler before the - # handshake frame is even written, so the initialize commits first. await entered.wait() - init = await client.send_raw_request("initialize", _initialize_params()) - assert init["protocolVersion"] == LATEST_HANDSHAKE_VERSION - release.set() - assert modern_result["tools"][0]["name"] == "t" - # The straggler's success did not move the era: plain legacy requests - # still serve (a modern overwrite would demand the envelope triple). - result = await client.send_raw_request("tools/list", None) - assert result["tools"][0]["name"] == "t" - - -@pytest.mark.anyio -async def test_dual_era_loop_modern_success_cancelled_away_at_the_response_write_never_locks(): - """A peer cancel that lands while the handler is finishing means the - dispatcher replaces the computed result with "Request cancelled" - the - client never sees the success, so the era must not lock and the legacy - handshake must stay available.""" - entered = anyio.Event() - release = anyio.Event() - - async def list_tools(ctx: Ctx, params: PaginatedRequestParams | None) -> ListToolsResult: - entered.set() - # Survive the interrupt-mode scope cancel so the handler completes - # with the cancel pending - the cancellation is then delivered at the - # dispatcher's response-write checkpoint, after the era commit ran. - with anyio.CancelScope(shield=True): - await release.wait() - return ListToolsResult(tools=[]) - - parked = Server(name="parked-server", version="0.0.1", on_list_tools=list_tools) - async with dual_era_client(parked) as (client, _): - failures: list[MCPError] = [] - - async def modern_call() -> None: with pytest.raises(MCPError) as exc_info: - await client.send_raw_request("tools/list", _modern_params()) - failures.append(exc_info.value) - - async with anyio.create_task_group() as tg: - tg.start_soon(modern_call) - await entered.wait() - # First request on a fresh dispatcher pair, so its id is 1. - await client.notify("notifications/cancelled", {"requestId": 1}) - # The read loop handles frames in order: this marker's response - # proves the cancel was processed before the handler resumes. - with pytest.raises(MCPError): - await client.send_raw_request("probe/marker", None) + await client.send_raw_request("initialize", _initialize_params()) + assert exc_info.value.error.code == UNSUPPORTED_PROTOCOL_VERSION release.set() - assert failures[0].error.message == "Request cancelled" - # The cancelled-away success never locked the era. - init = await client.send_raw_request("initialize", _initialize_params()) - assert init["protocolVersion"] == LATEST_HANDSHAKE_VERSION + assert modern_result["tools"][0]["name"] == "t" @pytest.mark.anyio @@ -1857,8 +1824,7 @@ async def test_dual_era_loop_maps_unmapped_handler_exceptions_like_the_modern_ht """An unmapped handler exception on a modern request surfaces as the generic INTERNAL_ERROR - the same boundary as the modern HTTP entry - so handler internals never reach the wire. (The dispatcher's code-0 - catch-all is a handshake-era compat pin and stays legacy-only.) The - failed request never locks, so the handshake stays available.""" + catch-all is a handshake-era compat pin and stays legacy-only.)""" async def list_tools(ctx: Ctx, params: PaginatedRequestParams | None) -> ListToolsResult: raise RuntimeError("handler internals") @@ -1867,8 +1833,6 @@ async def list_tools(ctx: Ctx, params: PaginatedRequestParams | None) -> ListToo async with dual_era_client(exploding) as (client, _): with pytest.raises(MCPError) as exc_info: await client.send_raw_request("tools/list", _modern_params()) - init = await client.send_raw_request("initialize", _initialize_params()) - assert init["protocolVersion"] == LATEST_HANDSHAKE_VERSION assert exc_info.value.error.code == INTERNAL_ERROR assert exc_info.value.error.message == "Internal server error" assert "handler internals" not in str(exc_info.value.error) From c3b0e01ff2f3f8f910275fdd1bab762bb57efb97 Mon Sep 17 00:00:00 2001 From: Max Isbey <224885523+maxisbey@users.noreply.github.com> Date: Thu, 23 Jul 2026 15:29:12 +0000 Subject: [PATCH 2/4] Address review: keep sender contexts through the replay, bound the opening peek - The opening-request replay now carries each frame's captured sender context, so the streams the SSE transport hands the loop keep their contextvars propagation; pinned by a test on the dual-era loop. - Bound how many frames may precede the client's first request; past the limit the loop stops looking for an opening request and serves a handshake connection, so a peer streaming pre-request frames cannot grow the buffer without bound. - Drop the stale docs admonition that stdio still rejects listen, and trim the era-decision docstring to the rule it now implements. --- docs/handlers/subscriptions.md | 7 ----- src/mcp/server/runner.py | 52 ++++++++++++++++++++++------------ tests/server/test_runner.py | 48 +++++++++++++++++++++++++++++++ 3 files changed, 82 insertions(+), 25 deletions(-) diff --git a/docs/handlers/subscriptions.md b/docs/handlers/subscriptions.md index 85b963278..99b1bc98a 100644 --- a/docs/handlers/subscriptions.md +++ b/docs/handlers/subscriptions.md @@ -52,13 +52,6 @@ Two things the stream is *not*: handler on the low-level `Server` and acknowledge a smaller filter than the client asked for; the acknowledgment is how the client learns what it actually got. -!!! warning "Streamable HTTP only, for now" - `subscriptions/listen` needs a transport that can stream a request's response, which today - means streamable HTTP. Over stdio a 2026-07-28 connection rejects the method with - METHOD_NOT_FOUND, even though `server/discover` advertises the subscription capabilities - there. Serving it over stdio is planned; the open-stream semantics for that transport are - not built yet. - ## The client end Here is a client on the other side of that stream, following the board: diff --git a/src/mcp/server/runner.py b/src/mcp/server/runner.py index 5e80a1dcd..3f3d29c10 100644 --- a/src/mcp/server/runner.py +++ b/src/mcp/server/runner.py @@ -13,6 +13,7 @@ from __future__ import annotations +import contextvars import logging from collections.abc import AsyncIterator, Awaitable, Mapping from contextlib import asynccontextmanager @@ -59,6 +60,7 @@ from mcp.server.context import CallNext, HandlerResult, ServerMiddleware, ServerRequestContext from mcp.server.models import InitializationOptions from mcp.server.session import ServerSession +from mcp.shared._context_streams import ContextReceiveStream from mcp.shared._stream_protocols import ReadStream, WriteStream from mcp.shared.dispatcher import CallOptions, DispatchContext, Dispatcher, OnNotify, OnRequest from mcp.shared.exceptions import MCPError, NoBackChannelError @@ -499,15 +501,12 @@ async def serve_loop( def _has_modern_envelope(params: Mapping[str, Any] | None) -> bool: """Whether `params._meta` carries the reserved protocol-version key. - Era evidence is the client's explicit version declaration: the - `io.modelcontextprotocol/protocolVersion` key exists only in 2026-07-28+ - envelopes, and the `io.modelcontextprotocol/` prefix is spec-reserved, so - legacy traffic never mints it (bare `_meta` is NOT evidence - legacy - requests carry `progressToken` there). Presence of the version key alone - is the rule, not the full required pair, so a half-built envelope - (version present, capabilities missing) still routes modern and gets the - classifier's INVALID_PARAMS naming the missing key instead of the legacy - path's generic one - and, like every failed classification, locks no era. + The `io.modelcontextprotocol/protocolVersion` key exists only in + 2026-07-28+ envelopes and its prefix is spec-reserved, so legacy traffic + never mints it (a bare `_meta` is not evidence - legacy requests carry + `progressToken` there). The version key alone is the signal, not the full + required pair, so a half-built envelope still routes modern and gets the + classifier's INVALID_PARAMS naming the missing key. """ if not params: return False @@ -644,6 +643,17 @@ async def serve_dual_era_loop( await write_stream.aclose() +_OPENING_PEEK_LIMIT: int = 32 +"""How many frames may precede the client's first request before the loop +stops looking for one. Every legitimate client opens with a request, so this +only bounds a peer that streams other frames without ever sending one.""" + + +def _sender_context(stream: ReadStream[Any]) -> contextvars.Context: + """The per-message sender context a context-aware stream carries, else the current one.""" + return getattr(stream, "last_context", None) or contextvars.copy_context() + + @asynccontextmanager async def _replay_from_opening_request( read_stream: ReadStream[SessionMessage | Exception], @@ -652,29 +662,35 @@ async def _replay_from_opening_request( Reads frames until the first JSON-RPC request arrives, then yields that request together with a stream that replays every frame read so far and - relays the rest of `read_stream` behind it. The request is `None` if the - client closed the channel before sending any request. + relays the rest of `read_stream` behind it, sender contexts included. The + request is `None` if the channel closes - or `_OPENING_PEEK_LIMIT` frames + pass - before any request appears. """ - peeked: list[SessionMessage | Exception] = [] + peeked: list[tuple[contextvars.Context, SessionMessage | Exception]] = [] opening: JSONRPCRequest | None = None - replay_send, replayed = anyio.create_memory_object_stream[SessionMessage | Exception]() + replay_send, replay_receive = anyio.create_memory_object_stream[ + tuple[contextvars.Context, SessionMessage | Exception] + ]() + replayed = ContextReceiveStream(replay_receive) async def replay_then_relay() -> None: async with replay_send: - for item in peeked: - await replay_send.send(item) + for envelope in peeked: + await replay_send.send(envelope) async for item in read_stream: - await replay_send.send(item) + await replay_send.send((_sender_context(read_stream), item)) # This helper takes ownership of `read_stream` from the serving loop, so # every exit - including cancellation while awaiting the first request - # closes it and the replay channel. try: async for item in read_stream: - peeked.append(item) + peeked.append((_sender_context(read_stream), item)) if isinstance(item, SessionMessage) and isinstance(item.message, JSONRPCRequest): opening = item.message break + if len(peeked) >= _OPENING_PEEK_LIMIT: + break async with anyio.create_task_group() as tg: tg.start_soon(replay_then_relay) yield opening, replayed @@ -682,7 +698,7 @@ async def replay_then_relay() -> None: finally: await read_stream.aclose() replay_send.close() - replayed.close() + replay_receive.close() async def _serve_legacy_stream( diff --git a/tests/server/test_runner.py b/tests/server/test_runner.py index 484593a3d..169e13993 100644 --- a/tests/server/test_runner.py +++ b/tests/server/test_runner.py @@ -7,6 +7,7 @@ `aclose_shielded`) follow at the bottom. """ +import contextvars from collections.abc import AsyncIterator, Mapping from contextlib import asynccontextmanager from dataclasses import dataclass, field, replace @@ -57,6 +58,7 @@ from mcp.server.lowlevel.server import NotificationOptions, Server from mcp.server.models import InitializationOptions from mcp.server.runner import ( + _OPENING_PEEK_LIMIT, ServerRunner, _extract_meta, _has_modern_envelope, @@ -69,6 +71,7 @@ ) from mcp.server.session import ServerSession from mcp.server.subscriptions import SUBSCRIPTION_ID_META_KEY, InMemorySubscriptionBus, ListenHandler +from mcp.shared._context_streams import create_context_streams from mcp.shared.dispatcher import CallOptions from mcp.shared.exceptions import MCPError, NoBackChannelError from mcp.shared.jsonrpc_dispatcher import JSONRPCDispatcher @@ -1356,6 +1359,8 @@ async def _append_async(dst: list[int], v: int) -> None: _LIFESPAN: dict[str, Any] = {} +_SENDER_VAR: contextvars.ContextVar[str] = contextvars.ContextVar("dual_era_sender_var", default="unset") + @pytest.mark.anyio async def test_serve_one_runs_handler_and_returns_result_dict(server: SrvT): @@ -1972,6 +1977,49 @@ async def test_notify_only_outbound_forwards_notifications_and_refuses_requests( await outbound.send_raw_request("ping", None) +@pytest.mark.anyio +async def test_dual_era_loop_carries_the_sender_context_through_the_replay(): + """A context-aware read stream's per-message sender context still reaches + handlers over the dual-era loop (the SSE transport delivers requests this + way): the opening-request replay forwards each frame's captured context.""" + seen: list[str] = [] + + async def probe(ctx: Ctx, params: RequestParams | None) -> dict[str, Any]: + seen.append(_SENDER_VAR.get()) + return {} + + server = Server(name="ctx-server", version="0.0.1") + server.add_request_handler("x/probe", RequestParams, probe) + c2s_send, c2s_recv = create_context_streams[SessionMessage | Exception](8) + s2c_send, s2c_recv = anyio.create_memory_object_stream[SessionMessage | Exception](8) + frame = JSONRPCRequest(jsonrpc="2.0", id=1, method="x/probe", params=_modern_params()) + async with anyio.create_task_group() as tg, s2c_recv: + tg.start_soon(partial(serve_dual_era_loop, server, c2s_recv, s2c_send, lifespan_state=_LIFESPAN)) + token = _SENDER_VAR.set("from-the-sender") + try: + await c2s_send.send(SessionMessage(message=frame)) + finally: + _SENDER_VAR.reset(token) + with anyio.fail_after(5): + await s2c_recv.receive() + tg.cancel_scope.cancel() + await c2s_send.aclose() + assert seen == ["from-the-sender"] + + +@pytest.mark.anyio +async def test_dual_era_loop_stops_peeking_after_a_flood_of_pre_request_frames(server: SrvT): + """A peer that streams notifications and never opens with a request only + gets the first `_OPENING_PEEK_LIMIT` frames buffered: past that the loop + stops looking for an opening request and serves the rest as an ordinary + handshake connection, so the handshake that follows still lands.""" + async with dual_era_client(server) as (client, _): + for _ in range(_OPENING_PEEK_LIMIT + 5): + await client.notify("notifications/flood", None) + init = await client.send_raw_request("initialize", _initialize_params()) + assert init["protocolVersion"] == LATEST_HANDSHAKE_VERSION + + @pytest.mark.anyio async def test_dual_era_client_propagates_body_exception_unwrapped(server: SrvT): """The harness re-raises body exceptions as-is, not as `ExceptionGroup`.""" From 46c04ba23fd9f44b817d085a0b0fd3665a904593 Mon Sep 17 00:00:00 2001 From: Max Isbey <224885523+maxisbey@users.noreply.github.com> Date: Thu, 23 Jul 2026 17:29:41 +0000 Subject: [PATCH 3/4] Address review round two: bounded lead, exact context check, prose - Frames arriving ahead of the client's first request are kept only up to a small bound and never decide the era: past the bound they are dropped while the loop keeps waiting for the opening request, so a bare notifications/initialized still opens the gate and a flood cannot grow the buffer or pick an era. - Check the captured sender context with `is not None`, since an empty contextvars.Context is falsy. - Word the legacy refusal to the actual rule (first request carried no 2026 envelope) and update the Server.run docstring and one test to the decide-from-the-opening-request model. --- src/mcp/server/lowlevel/server.py | 6 ++--- src/mcp/server/runner.py | 44 ++++++++++++++++--------------- tests/server/test_runner.py | 27 +++++++++---------- 3 files changed, 39 insertions(+), 38 deletions(-) diff --git a/src/mcp/server/lowlevel/server.py b/src/mcp/server/lowlevel/server.py index 5088c9278..1cbd3f2bd 100644 --- a/src/mcp/server/lowlevel/server.py +++ b/src/mcp/server/lowlevel/server.py @@ -703,9 +703,9 @@ 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 to succeed - locks the connection). Transports with their own lifespan owner (the - streamable-HTTP manager) call `serve_loop` directly instead. + per-request-envelope era (the client's first request decides which). + Transports with their own lifespan owner (the streamable-HTTP manager) + call `serve_loop` directly instead. """ async with self.lifespan(self) as lifespan_context: await serve_dual_era_loop( diff --git a/src/mcp/server/runner.py b/src/mcp/server/runner.py index 3f3d29c10..86e866a53 100644 --- a/src/mcp/server/runner.py +++ b/src/mcp/server/runner.py @@ -643,31 +643,31 @@ async def serve_dual_era_loop( await write_stream.aclose() -_OPENING_PEEK_LIMIT: int = 32 -"""How many frames may precede the client's first request before the loop -stops looking for one. Every legitimate client opens with a request, so this -only bounds a peer that streams other frames without ever sending one.""" +_PRE_REQUEST_REPLAY_LIMIT: int = 8 +"""How many frames arriving ahead of the client's first request are kept +for the chosen era's loop (a bare `notifications/initialized` is the one that +matters); further ones are dropped and never decide the era.""" def _sender_context(stream: ReadStream[Any]) -> contextvars.Context: """The per-message sender context a context-aware stream carries, else the current one.""" - return getattr(stream, "last_context", None) or contextvars.copy_context() + ctx = getattr(stream, "last_context", None) + return ctx if ctx is not None else contextvars.copy_context() @asynccontextmanager async def _replay_from_opening_request( read_stream: ReadStream[SessionMessage | Exception], ) -> AsyncIterator[tuple[JSONRPCRequest | None, ReadStream[SessionMessage | Exception]]]: - """Peek at the client's first request without consuming anything. + """Peek at the client's first request without consuming it. - Reads frames until the first JSON-RPC request arrives, then yields that - request together with a stream that replays every frame read so far and - relays the rest of `read_stream` behind it, sender contexts included. The - request is `None` if the channel closes - or `_OPENING_PEEK_LIMIT` frames - pass - before any request appears. + Yields that request together with a stream that replays it - preceded by + up to `_PRE_REQUEST_REPLAY_LIMIT` earlier frames - and relays the rest of + `read_stream` behind it, sender contexts included. The request is `None` + if the channel closes before one arrives. """ - peeked: list[tuple[contextvars.Context, SessionMessage | Exception]] = [] - opening: JSONRPCRequest | None = None + lead: list[tuple[contextvars.Context, SessionMessage | Exception]] = [] + opening_request: JSONRPCRequest | None = None replay_send, replay_receive = anyio.create_memory_object_stream[ tuple[contextvars.Context, SessionMessage | Exception] ]() @@ -675,7 +675,7 @@ async def _replay_from_opening_request( async def replay_then_relay() -> None: async with replay_send: - for envelope in peeked: + for envelope in lead: await replay_send.send(envelope) async for item in read_stream: await replay_send.send((_sender_context(read_stream), item)) @@ -685,15 +685,17 @@ async def replay_then_relay() -> None: # closes it and the replay channel. try: async for item in read_stream: - peeked.append((_sender_context(read_stream), item)) if isinstance(item, SessionMessage) and isinstance(item.message, JSONRPCRequest): - opening = item.message - break - if len(peeked) >= _OPENING_PEEK_LIMIT: + opening_request = item.message + elif len(lead) >= _PRE_REQUEST_REPLAY_LIMIT: + logger.debug("dropped a frame received before the first request: %r", item) + continue + lead.append((_sender_context(read_stream), item)) + if opening_request is not None: break async with anyio.create_task_group() as tg: tg.start_soon(replay_then_relay) - yield opening, replayed + yield opening_request, replayed tg.cancel_scope.cancel() finally: await read_stream.aclose() @@ -728,8 +730,8 @@ async def on_request( if method != "initialize" and _has_modern_envelope(params): raise MCPError( code=INVALID_REQUEST, - message="connection was opened with the initialize handshake; " - "2026-07-28 envelope requests are not accepted on it", + message="this connection's first request carried no 2026-07-28 envelope, so it " + "serves the handshake era; enveloped requests are not accepted on it", ) return await runner.on_request(dctx, method, params) diff --git a/tests/server/test_runner.py b/tests/server/test_runner.py index 169e13993..0607be3ba 100644 --- a/tests/server/test_runner.py +++ b/tests/server/test_runner.py @@ -58,7 +58,6 @@ from mcp.server.lowlevel.server import NotificationOptions, Server from mcp.server.models import InitializationOptions from mcp.server.runner import ( - _OPENING_PEEK_LIMIT, ServerRunner, _extract_meta, _has_modern_envelope, @@ -1545,7 +1544,7 @@ async def test_dual_era_loop_initialize_locks_legacy_and_rejects_modern_traffic( assert result["tools"][0]["name"] == "t" assert discover_exc.value.error.code == INVALID_REQUEST assert envelope_exc.value.error.code == INVALID_REQUEST - assert "opened with the initialize handshake" in envelope_exc.value.error.message + assert "carried no 2026-07-28 envelope" in envelope_exc.value.error.message @pytest.mark.anyio @@ -1725,9 +1724,9 @@ async def test_dual_era_loop_initialize_with_envelope_takes_the_handshake_path(s @pytest.mark.anyio -async def test_dual_era_loop_modern_notification_dispatches_at_locked_version(server: SrvT): - """Notifications carry no envelope, so on a modern-locked connection they - dispatch with the locked protocol version.""" +async def test_dual_era_loop_modern_notification_dispatches_at_the_served_version(server: SrvT): + """Notifications carry no envelope, so on a 2026 connection they dispatch + at the modern version the server serves.""" seen_versions: list[str] = [] handled = anyio.Event() @@ -2008,16 +2007,16 @@ async def probe(ctx: Ctx, params: RequestParams | None) -> dict[str, Any]: @pytest.mark.anyio -async def test_dual_era_loop_stops_peeking_after_a_flood_of_pre_request_frames(server: SrvT): - """A peer that streams notifications and never opens with a request only - gets the first `_OPENING_PEEK_LIMIT` frames buffered: past that the loop - stops looking for an opening request and serves the rest as an ordinary - handshake connection, so the handshake that follows still lands.""" +async def test_dual_era_loop_leading_notifications_never_decide_the_era(server: SrvT): + """Frames a peer sends ahead of its first request never decide the era: a + stream of leading notifications well past the retained lead is followed by + an enveloped request, which still opens a 2026 connection and serves. The + lead is bounded, so the flood cannot grow the loop's buffer either.""" async with dual_era_client(server) as (client, _): - for _ in range(_OPENING_PEEK_LIMIT + 5): - await client.notify("notifications/flood", None) - init = await client.send_raw_request("initialize", _initialize_params()) - assert init["protocolVersion"] == LATEST_HANDSHAKE_VERSION + for _ in range(50): + await client.notify("notifications/lead", None) + result = await client.send_raw_request("tools/list", _modern_params()) + assert result["tools"][0]["name"] == "t" @pytest.mark.anyio From a0d62336f2aa559ad47b44f0bc076edddbf5e04f Mon Sep 17 00:00:00 2001 From: Max Isbey <224885523+maxisbey@users.noreply.github.com> Date: Thu, 23 Jul 2026 19:12:52 +0000 Subject: [PATCH 4/4] Address review round three: EOF tolerance, docstring, refusal wording - Treat a read stream whose receive end the transport closed as end-of-input in the opening peek and the relay, mirroring the tolerance the dispatcher's read loop already had; the read loop moved here, so its EOF handling moves with it. Two tests close the stream before and after the first request and expect a clean return. - ListenHandler's docstring no longer claims it needs SSE mode. - The legacy refusal states the era rule instead of narrating what the first request was, so it holds for every opener. --- src/mcp/server/runner.py | 34 ++++++++++++++++++++------------- src/mcp/server/subscriptions.py | 4 ++-- tests/server/test_runner.py | 33 +++++++++++++++++++++++++++++++- 3 files changed, 55 insertions(+), 16 deletions(-) diff --git a/src/mcp/server/runner.py b/src/mcp/server/runner.py index 86e866a53..6f9f7a8f7 100644 --- a/src/mcp/server/runner.py +++ b/src/mcp/server/runner.py @@ -677,22 +677,30 @@ async def replay_then_relay() -> None: async with replay_send: for envelope in lead: await replay_send.send(envelope) - async for item in read_stream: - await replay_send.send((_sender_context(read_stream), item)) + try: + async for item in read_stream: + await replay_send.send((_sender_context(read_stream), item)) + except anyio.ClosedResourceError: + # Receive end closed under us (stateless SHTTP teardown); same as EOF. + logger.debug("read stream closed by transport; treating as EOF") # This helper takes ownership of `read_stream` from the serving loop, so # every exit - including cancellation while awaiting the first request - # closes it and the replay channel. try: - async for item in read_stream: - if isinstance(item, SessionMessage) and isinstance(item.message, JSONRPCRequest): - opening_request = item.message - elif len(lead) >= _PRE_REQUEST_REPLAY_LIMIT: - logger.debug("dropped a frame received before the first request: %r", item) - continue - lead.append((_sender_context(read_stream), item)) - if opening_request is not None: - break + try: + async for item in read_stream: + if isinstance(item, SessionMessage) and isinstance(item.message, JSONRPCRequest): + opening_request = item.message + elif len(lead) >= _PRE_REQUEST_REPLAY_LIMIT: + logger.debug("dropped a frame received before the first request: %r", item) + continue + lead.append((_sender_context(read_stream), item)) + if opening_request is not None: + break + except anyio.ClosedResourceError: + # Receive end closed under us (stateless SHTTP teardown); same as EOF. + logger.debug("read stream closed by transport; treating as EOF") async with anyio.create_task_group() as tg: tg.start_soon(replay_then_relay) yield opening_request, replayed @@ -730,8 +738,8 @@ async def on_request( if method != "initialize" and _has_modern_envelope(params): raise MCPError( code=INVALID_REQUEST, - message="this connection's first request carried no 2026-07-28 envelope, so it " - "serves the handshake era; enveloped requests are not accepted on it", + message="this connection serves the handshake protocol era; " + "requests carrying the 2026-07-28 envelope are not accepted on it", ) return await runner.on_request(dctx, method, params) diff --git a/src/mcp/server/subscriptions.py b/src/mcp/server/subscriptions.py index 6b0b3d49b..f5d3b5b0a 100644 --- a/src/mcp/server/subscriptions.py +++ b/src/mcp/server/subscriptions.py @@ -164,8 +164,8 @@ class ListenHandler: cancels the handler; the stream just ends, per the spec's abrupt-close contract) or `close` ends all streams gracefully. - Requires a transport that can stream a request's response (streamable - HTTP's SSE mode). + Served on any transport that can carry the request's response stream: + streamable HTTP's SSE mode, or a duplex stream pair such as stdio. `max_subscriptions` bounds concurrent streams (further listen requests are rejected with `INTERNAL_ERROR`, before the ack). `max_buffered_events` diff --git a/tests/server/test_runner.py b/tests/server/test_runner.py index 0607be3ba..eb212dafd 100644 --- a/tests/server/test_runner.py +++ b/tests/server/test_runner.py @@ -16,6 +16,7 @@ import anyio import anyio.abc +import anyio.lowlevel import pytest from mcp_types import ( CLIENT_CAPABILITIES_META_KEY, @@ -1544,7 +1545,7 @@ async def test_dual_era_loop_initialize_locks_legacy_and_rejects_modern_traffic( assert result["tools"][0]["name"] == "t" assert discover_exc.value.error.code == INVALID_REQUEST assert envelope_exc.value.error.code == INVALID_REQUEST - assert "carried no 2026-07-28 envelope" in envelope_exc.value.error.message + assert "serves the handshake protocol era" in envelope_exc.value.error.message @pytest.mark.anyio @@ -2019,6 +2020,36 @@ async def test_dual_era_loop_leading_notifications_never_decide_the_era(server: assert result["tools"][0]["name"] == "t" +@pytest.mark.anyio +async def test_dual_era_loop_treats_a_read_stream_closed_before_the_first_request_as_end_of_input(server: SrvT): + """A transport closing the read stream's receive end (the stateless teardown + pattern) ends the loop like end-of-input rather than surfacing a + closed-resource error - here before the client has sent anything.""" + c2s_send, c2s_recv = anyio.create_memory_object_stream[SessionMessage | Exception](8) + s2c_send, s2c_recv = anyio.create_memory_object_stream[SessionMessage | Exception](8) + with anyio.fail_after(5): + async with anyio.create_task_group() as tg, c2s_send, s2c_recv: + tg.start_soon(partial(serve_dual_era_loop, server, c2s_recv, s2c_send, lifespan_state=_LIFESPAN)) + await anyio.lowlevel.checkpoint() + c2s_recv.close() # the transport tears down under the waiting loop + + +@pytest.mark.anyio +async def test_dual_era_loop_treats_a_read_stream_closed_mid_connection_as_end_of_input(server: SrvT): + """The same teardown after the connection is open ends the era loop + cleanly: the relay behind the opening request treats the closed receive + end as end-of-input.""" + c2s_send, c2s_recv = anyio.create_memory_object_stream[SessionMessage | Exception](8) + s2c_send, s2c_recv = anyio.create_memory_object_stream[SessionMessage | Exception](8) + ping = JSONRPCRequest(jsonrpc="2.0", id=1, method="ping", params=None) + with anyio.fail_after(5): + async with anyio.create_task_group() as tg, c2s_send, s2c_recv: + tg.start_soon(partial(serve_dual_era_loop, server, c2s_recv, s2c_send, lifespan_state=_LIFESPAN)) + await c2s_send.send(SessionMessage(message=ping)) + await s2c_recv.receive() # the connection is open and answered + c2s_recv.close() + + @pytest.mark.anyio async def test_dual_era_client_propagates_body_exception_unwrapped(server: SrvT): """The harness re-raises body exceptions as-is, not as `ExceptionGroup`."""