diff --git a/docs/handlers/subscriptions.md b/docs/handlers/subscriptions.md index 85b9632786..99b1bc98ad 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/docs/whats-new.md b/docs/whats-new.md index 9c7a9be911..3f1188f8fc 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/lowlevel/server.py b/src/mcp/server/lowlevel/server.py index 5088c92788..1cbd3f2bd6 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 98bc133a7e..6f9f7a8f74 100644 --- a/src/mcp/server/runner.py +++ b/src/mcp/server/runner.py @@ -13,11 +13,13 @@ from __future__ import annotations +import contextvars 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 +39,7 @@ Implementation, InitializeRequestParams, InitializeResult, + JSONRPCRequest, RequestId, RequestParams, RequestParamsMeta, @@ -57,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 @@ -497,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 @@ -605,91 +606,180 @@ 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() + + +_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.""" + 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 it. + + 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. + """ + 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] + ]() + replayed = ContextReceiveStream(replay_receive) + + async def replay_then_relay() -> None: + async with replay_send: + for envelope in lead: + await replay_send.send(envelope) + 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: + 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 + tg.cancel_scope.cancel() + finally: + await read_stream.aclose() + replay_send.close() + replay_receive.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="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) + + 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 +788,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/src/mcp/server/subscriptions.py b/src/mcp/server/subscriptions.py index 6b0b3d49b5..f5d3b5b0a4 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 174cb853ae..eb212dafdb 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 @@ -15,6 +16,7 @@ import anyio import anyio.abc +import anyio.lowlevel import pytest from mcp_types import ( CLIENT_CAPABILITIES_META_KEY, @@ -68,6 +70,8 @@ serve_one, ) 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 @@ -1355,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): @@ -1539,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 "locked to the legacy handshake era" in discover_exc.value.error.message + assert "serves the handshake protocol era" in envelope_exc.value.error.message @pytest.mark.anyio @@ -1558,44 +1564,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 +1621,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 +1681,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 @@ -1705,9 +1725,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() @@ -1774,11 +1794,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 +1816,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 +1829,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 +1838,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) @@ -2008,6 +1977,79 @@ 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_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(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 +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`."""