From ff34da9147baaa3e249ea28df2d5fb4998f24923 Mon Sep 17 00:00:00 2001 From: Max Isbey <224885523+maxisbey@users.noreply.github.com> Date: Thu, 16 Jul 2026 21:22:04 +0000 Subject: [PATCH 1/7] Add per-request cancel-answer and success-frame seams to JSONRPCDispatcher Two per-request facts on the dispatch context, both defaulting to the existing behavior: - cancel_answer: the error written when a peer cancel interrupts the handler. Defaults to the legacy code-0 "Request cancelled" compat answer; None means total silence (the 2026 rule that nothing follows notifications/cancelled). Handler-origin error responses now also honor the silence commitment, so an error computed after the cancel landed (shielded handler, synchronous raise) never reaches the wire. - on_success_frame: an observer invoked immediately before each non-error frame written for the request (request-scoped notifications, progress, the success result), with no checkpoint before the write. Loop layers use it to commit connection state no later than the request's first client-visible output. Configured via suppress_cancel_answer / observe_success_frames, which no-op on dispatch contexts that are already structurally silent after a cancel (single-exchange HTTP, direct dispatch). The dispatcher stays era-ignorant; the era knowledge lives in whoever calls the setters. --- src/mcp/shared/jsonrpc_dispatcher.py | 96 +++++++++++++-- tests/shared/test_jsonrpc_dispatcher.py | 154 +++++++++++++++++++++++- 2 files changed, 242 insertions(+), 8 deletions(-) diff --git a/src/mcp/shared/jsonrpc_dispatcher.py b/src/mcp/shared/jsonrpc_dispatcher.py index 42798fdc54..1f3b29479f 100644 --- a/src/mcp/shared/jsonrpc_dispatcher.py +++ b/src/mcp/shared/jsonrpc_dispatcher.py @@ -12,7 +12,7 @@ from collections.abc import Awaitable, Callable, Mapping from dataclasses import dataclass, field from functools import partial -from typing import Any, Generic, Literal, cast +from typing import Any, Final, Generic, Literal, cast import anyio import anyio.abc @@ -64,7 +64,9 @@ "JSONRPCDispatcher", "cancelled_request_id_from_params", "handler_exception_to_error_data", + "observe_success_frames", "progress_token_from_params", + "suppress_cancel_answer", ] logger = logging.getLogger(__name__) @@ -131,6 +133,15 @@ class _InFlight(Generic[TransportT]): dctx: _JSONRPCDispatchContext[TransportT] +_LEGACY_CANCEL_ANSWER: Final[ErrorData] = ErrorData(code=0, message="Request cancelled") +"""The answer a peer-cancelled request gets by default. + +TODO(L38): the spec says receivers SHOULD NOT respond after a cancel; the +existing server always has, so this pins released-client compat. Never +mutated - one shared instance is safe. +""" + + @dataclass class _JSONRPCDispatchContext(Generic[TransportT]): """Concrete `DispatchContext` produced for each inbound JSON-RPC message.""" @@ -143,6 +154,22 @@ class _JSONRPCDispatchContext(Generic[TransportT]): _progress_token: ProgressToken | None = None _closed: bool = False cancel_requested: anyio.Event = field(default_factory=anyio.Event) + cancel_answer: ErrorData | None = field(default_factory=lambda: _LEGACY_CANCEL_ANSWER) + """The error written when a peer cancel interrupts this request's handler. + + `None` means silence: no frame at all follows the cancel. The default is + the legacy compat answer; loop layers that know a request is governed by + 2026-era wire rules (which forbid any frame for a request after + `notifications/cancelled`) clear it via `suppress_cancel_answer`. + """ + on_success_frame: Callable[[], None] | None = None + """Observer invoked synchronously immediately before each non-error frame + written for this request: a request-scoped notification (progress rides + those) or the success result. Never invoked for error responses, nor for + notifications dropped because the context closed. Set via + `observe_success_frames`; loop layers use it to commit connection state no + later than the request's first client-visible output. + """ @property def request_id(self) -> RequestId | None: @@ -156,6 +183,8 @@ async def notify(self, method: str, params: Mapping[str, Any] | None, opts: Call if self._closed: logger.debug("dropped %s: dispatch context closed", method) return + if self.on_success_frame is not None: + self.on_success_frame() await self._dispatcher.notify(method, params, opts, _related_request_id=self._request_id) async def send_raw_request( @@ -182,6 +211,41 @@ def close(self) -> None: self._closed = True +def suppress_cancel_answer(dctx: DispatchContext[Any]) -> None: + """Commit `dctx`'s request to silence on peer cancellation. + + 2026-era wire rule: once the peer sends `notifications/cancelled` for a + request, the server MUST NOT send any further frame for it - no result, + no error. Era-aware loop layers call this the moment a request commits to + those semantics; do so before the request's first checkpoint, so a racing + cancel can never observe the legacy default answer. + + No-op for dispatch contexts that are already structurally silent after a + cancel (single-exchange HTTP closes the response stream; direct dispatch + unwinds into the caller's own cancelled scope) - only the JSON-RPC loop + context writes a late answer to suppress. + """ + if isinstance(dctx, _JSONRPCDispatchContext): + dctx.cancel_answer = None + + +def observe_success_frames(dctx: DispatchContext[Any], observer: Callable[[], None]) -> None: + """Invoke `observer` immediately before each non-error frame written for `dctx`'s request. + + Fires for request-scoped notifications (progress rides those) and for the + success result, in the writing task with no checkpoint before the write - + so state the observer commits is settled before the peer can read the + frame. It does not fire for error responses or for notifications dropped + because the request already finished. `observer` must not raise. + + No-op for dispatch contexts other than the JSON-RPC loop's: the only + caller is the dual-era loop's era lock, which has no analogue on the + single-exchange or direct paths. + """ + if isinstance(dctx, _JSONRPCDispatchContext): + dctx.on_success_frame = observer + + def _default_transport_builder(_meta: MessageMetadata) -> TransportContext: return TransportContext(kind="jsonrpc", can_send_request=True) @@ -717,15 +781,18 @@ async def _handle_request( # peers drop late responses, while a second answer for one id # would break JSON-RPC. answer_write_started = True + if dctx.on_success_frame is not None: + dctx.on_success_frame() await self._write_result(req.id, result) - if scope.cancelled_caught: + if scope.cancelled_caught and dctx.cancel_answer is not None: # anyio absorbs the scope's own cancel at __exit__, and # `cancelled_caught` (unlike `cancel_called`) guarantees the # result write above did not happen - no double response. - # TODO(L38): spec says SHOULD NOT respond after cancel; - # the existing server always has, so match that for now. + # `cancel_answer` is the legacy compat answer unless the loop + # layer committed this request to 2026 cancel semantics + # (silence) via `suppress_cancel_answer`. answer_write_started = True - await self._write_error(req.id, ErrorData(code=0, message="Request cancelled")) + await self._write_error(req.id, dctx.cancel_answer) except anyio.get_cancelled_exc_class(): # Shutdown: answer the request so the peer isn't left waiting - unless # an answer write already started (it may have reached the transport; @@ -742,12 +809,12 @@ async def _handle_request( except Exception as e: error = handler_exception_to_error_data(e) if error is not None: - await self._write_error(req.id, error) + await self._answer_error(dctx, req.id, error) else: logger.exception("handler for %r raised", req.method) # TODO(L58): code=0 pins existing-server compat; JSON-RPC says # INTERNAL_ERROR. Revisit per the suite's divergence entry. - await self._write_error(req.id, ErrorData(code=0, message=str(e))) + await self._answer_error(dctx, req.id, ErrorData(code=0, message=str(e))) if self._raise_handler_exceptions: raise # No `_in_flight` pop here: the inner finally covers every path, and a late pop could evict a reused id. @@ -771,6 +838,21 @@ async def _write_error(self, request_id: RequestId, error: ErrorData) -> None: except (anyio.BrokenResourceError, anyio.ClosedResourceError): logger.debug("dropped error for %r: write stream closed", request_id) + async def _answer_error( + self, dctx: _JSONRPCDispatchContext[TransportT], request_id: RequestId, error: ErrorData + ) -> None: + """Write a handler-origin error response, honoring the request's cancel-silence commitment. + + A request committed to silence via `suppress_cancel_answer` whose peer + has already cancelled gets NO frame at all - even when the handler's + failure was computed before the cancellation interrupt could land (a + synchronous raise after the cancel arrived, or a shielded handler). + """ + if dctx.cancel_answer is None and dctx.cancel_requested.is_set(): + logger.debug("dropped error for %r: peer cancelled a silence-committed request", request_id) + return + await self._write_error(request_id, error) + async def _final_write( self, write: Callable[[], Awaitable[None]], diff --git a/tests/shared/test_jsonrpc_dispatcher.py b/tests/shared/test_jsonrpc_dispatcher.py index 5c29c7e3ce..4deb6fccfb 100644 --- a/tests/shared/test_jsonrpc_dispatcher.py +++ b/tests/shared/test_jsonrpc_dispatcher.py @@ -41,11 +41,13 @@ _OutboundPlan, _Pending, _plan_outbound, + observe_success_frames, + suppress_cancel_answer, ) from mcp.shared.message import ClientMessageMetadata, MessageMetadata, ServerMessageMetadata, SessionMessage from mcp.shared.transport_context import TransportContext -from .conftest import jsonrpc_pair +from .conftest import direct_pair, jsonrpc_pair from .test_dispatcher import Recorder, echo_handlers, running_pair DCtx = DispatchContext[TransportContext] @@ -156,6 +158,156 @@ async def call_then_record() -> None: assert seen_error == [ErrorData(code=0, message="Request cancelled")] +@pytest.mark.anyio +async def test_suppress_cancel_answer_makes_a_peer_cancelled_request_fully_silent(): + """A request committed to 2026 cancel semantics gets NO late answer: after + the peer cancel nothing at all is written for its id (no result, no + error), and the loop keeps serving other requests.""" + c2s_send, c2s_recv = anyio.create_memory_object_stream[SessionMessage | Exception](32) + s2c_send, s2c_recv = anyio.create_memory_object_stream[SessionMessage | Exception](32) + server: JSONRPCDispatcher[TransportContext] = JSONRPCDispatcher(c2s_recv, s2c_send) + handler_started = anyio.Event() + handler_exited = anyio.Event() + + async def on_request(ctx: DCtx, method: str, params: Mapping[str, Any] | None) -> dict[str, Any]: + if method == "probe": + return {"alive": True} + suppress_cancel_answer(ctx) + handler_started.set() + try: + await anyio.sleep_forever() + finally: + handler_exited.set() + raise NotImplementedError + + async def on_notify(ctx: DCtx, method: str, params: Mapping[str, Any] | None) -> None: + pass + + try: + async with anyio.create_task_group() as tg: + await tg.start(server.run, on_request, on_notify) + with anyio.fail_after(5): + await c2s_send.send(SessionMessage(message=JSONRPCRequest(jsonrpc="2.0", id=7, method="slow"))) + await handler_started.wait() + await c2s_send.send( + SessionMessage( + message=JSONRPCNotification( + jsonrpc="2.0", method="notifications/cancelled", params={"requestId": 7} + ) + ) + ) + await handler_exited.wait() + # Once every task is parked again the cancelled handler task has + # fully unwound, so the buffered write stream would already + # carry any (buggy) late answer for id 7. + await anyio.wait_all_tasks_blocked() + await c2s_send.send(SessionMessage(message=JSONRPCRequest(jsonrpc="2.0", id=8, method="probe"))) + resp = await s2c_recv.receive() + assert isinstance(resp, SessionMessage) + assert isinstance(resp.message, JSONRPCResponse) + # The probe's answer is the FIRST frame the server ever wrote: + # total silence for the cancelled id. + assert resp.message.id == 8 + tg.cancel_scope.cancel() + finally: + for s in (c2s_send, c2s_recv, s2c_send, s2c_recv): + s.close() + + +@pytest.mark.anyio +async def test_suppress_cancel_answer_silences_even_an_error_raised_after_the_cancel_landed(): + """A handler that observes the peer cancel and raises before the + cancellation interrupt can land (shielded, then a synchronous raise) + still gets total silence: the error computed after the cancel must not + reach the wire either.""" + c2s_send, c2s_recv = anyio.create_memory_object_stream[SessionMessage | Exception](32) + s2c_send, s2c_recv = anyio.create_memory_object_stream[SessionMessage | Exception](32) + server: JSONRPCDispatcher[TransportContext] = JSONRPCDispatcher(c2s_recv, s2c_send) + handler_started = anyio.Event() + + async def on_request(ctx: DCtx, method: str, params: Mapping[str, Any] | None) -> dict[str, Any]: + if method == "probe": + return {"alive": True} + suppress_cancel_answer(ctx) + handler_started.set() + # Survive the interrupt so the cancel is observed, not delivered; the + # raise below then reaches the dispatcher's error arm directly. + with anyio.CancelScope(shield=True): + await ctx.cancel_requested.wait() + raise MCPError(code=INTERNAL_ERROR, message="computed after the cancel") + + async def on_notify(ctx: DCtx, method: str, params: Mapping[str, Any] | None) -> None: + pass + + try: + async with anyio.create_task_group() as tg: + await tg.start(server.run, on_request, on_notify) + with anyio.fail_after(5): + await c2s_send.send(SessionMessage(message=JSONRPCRequest(jsonrpc="2.0", id=7, method="slow"))) + await handler_started.wait() + await c2s_send.send( + SessionMessage( + message=JSONRPCNotification( + jsonrpc="2.0", method="notifications/cancelled", params={"requestId": 7} + ) + ) + ) + await anyio.wait_all_tasks_blocked() + await c2s_send.send(SessionMessage(message=JSONRPCRequest(jsonrpc="2.0", id=8, method="probe"))) + resp = await s2c_recv.receive() + assert isinstance(resp, SessionMessage) + assert isinstance(resp.message, JSONRPCResponse) + # The probe's answer is the first frame ever written: the + # post-cancel MCPError never reached the wire. + assert resp.message.id == 8 + tg.cancel_scope.cancel() + finally: + for s in (c2s_send, c2s_recv, s2c_send, s2c_recv): + s.close() + + +@pytest.mark.anyio +async def test_observe_success_frames_fires_for_notifications_and_the_result_but_not_errors(): + """The observer runs immediately before each non-error frame for the + request - the request-scoped notification and the success result - and + never for an error response.""" + calls: list[str] = [] + + async def on_request(ctx: DCtx, method: str, params: Mapping[str, Any] | None) -> dict[str, Any]: + observe_success_frames(ctx, lambda: calls.append(method)) + if method == "fail": + raise MCPError(code=INTERNAL_ERROR, message="nope") + await ctx.notify("x/note", None) + return {"ok": True} + + async with running_pair(jsonrpc_pair, server_on_request=on_request) as (client, *_): + with anyio.fail_after(5): + assert await client.send_raw_request("emit", None) == {"ok": True} + # Once before the notification frame, once before the result frame. + assert calls == ["emit", "emit"] + with pytest.raises(MCPError): + await client.send_raw_request("fail", None) + # Error responses never fire the observer. + assert calls == ["emit", "emit"] + + +@pytest.mark.anyio +async def test_cancel_and_frame_seams_are_noops_on_non_jsonrpc_contexts(): + """`suppress_cancel_answer` / `observe_success_frames` configure the + JSON-RPC loop context's late-answer machinery; other dispatch contexts + are already structurally silent after a cancel, so on the direct + dispatcher both are safe no-ops.""" + + async def on_request(ctx: DCtx, method: str, params: Mapping[str, Any] | None) -> dict[str, Any]: + suppress_cancel_answer(ctx) + observe_success_frames(ctx, lambda: None) + return {"ok": True} + + async with running_pair(direct_pair, server_on_request=on_request) as (client, *_): + with anyio.fail_after(5): + assert await client.send_raw_request("m", None) == {"ok": True} + + @pytest.mark.anyio async def test_peer_cancel_landing_after_handlers_last_checkpoint_writes_only_the_result(): """A peer cancel that fails to interrupt the handler writes only the result: one answer per From 8ca1fc628daaf4888fc779c6253f1adb98d3edb3 Mon Sep 17 00:00:00 2001 From: Max Isbey <224885523+maxisbey@users.noreply.github.com> Date: Thu, 16 Jul 2026 21:22:55 +0000 Subject: [PATCH 2/7] Serve subscriptions/listen over the dual-era stream loop Remove the transport-level rejection of subscriptions/listen on the stream-pair modern path: the method now dispatches through serve_one to the registered ListenHandler like every other modern request. Over stdio the listen request simply stays pending - the ack and every event are request-scoped notifications on the duplex pipe, and the response arrives only when the stream ends gracefully. server/discover's subscription capability advertisement is now truthful on this transport. Two era-scoped generalizations ride the dispatcher's new per-request seams, fixing the class rather than the listen instance: - Era lock at first client-visible success frame. A classified-modern request commits the modern lock immediately before its first non-error frame is written (notification, progress, or response), not when its dispatch returns. Plain requests lock at their response exactly as before; a listen locks at its ack, so an initialize pipelined behind a live stream is rejected with -32022 instead of flipping the connection legacy mid-stream. Classification failures and pre-frame cancellations still never lock, and progress written for a request that later fails locks correctly (the client that stamped a valid envelope already committed to modern). - Quiet cancellation for every classified-modern request. After an inbound notifications/cancelled the server sends nothing further for that request - no result, no error - matching the transport spec's MUST NOT. Legacy-era requests keep the byte-identical code-0 "Request cancelled" answer released 2025 clients unblock on. The single-exchange HTTP and direct in-process paths need no seam: cancellation is structurally silent there (stream close / caller's own scope). --- docs/handlers/subscriptions.md | 10 +- docs/whats-new.md | 2 +- src/mcp/server/runner.py | 75 ++++--- src/mcp/server/subscriptions.py | 5 +- tests/server/test_runner.py | 354 +++++++++++++++++++++++++++++--- 5 files changed, 384 insertions(+), 62 deletions(-) diff --git a/docs/handlers/subscriptions.md b/docs/handlers/subscriptions.md index 85b9632786..9e91b05e31 100644 --- a/docs/handlers/subscriptions.md +++ b/docs/handlers/subscriptions.md @@ -52,12 +52,10 @@ 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. +`subscriptions/listen` is served on every transport that can stream a request's response: over +streamable HTTP the response is the SSE stream, and over stdio the listen request simply stays +pending on the duplex pipe (the acknowledgment and every event are notifications; the request's +result arrives only when the stream ends gracefully). ## The client end diff --git a/docs/whats-new.md b/docs/whats-new.md index de29f0aefc..d619c4b8e7 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. This works on both transports: over streamable HTTP the stream is the request's SSE response, and over stdio the events ride the duplex pipe while the listen request stays pending. **[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 3b53335ae4..f03b39e408 100644 --- a/src/mcp/server/runner.py +++ b/src/mcp/server/runner.py @@ -59,7 +59,12 @@ from mcp.shared.dispatcher import CallOptions, DispatchContext, Dispatcher, OnNotify, OnRequest from mcp.shared.exceptions import MCPError, NoBackChannelError from mcp.shared.inbound import InboundLadderRejection, classify_inbound_request -from mcp.shared.jsonrpc_dispatcher import JSONRPCDispatcher, handler_exception_to_error_data +from mcp.shared.jsonrpc_dispatcher import ( + JSONRPCDispatcher, + handler_exception_to_error_data, + observe_success_frames, + suppress_cancel_answer, +) from mcp.shared.message import MessageMetadata, ServerMessageMetadata, SessionMessage from mcp.shared.transport_context import TransportContext @@ -575,15 +580,29 @@ async def serve_dual_era_loop( 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. + The modern lock commits on the FIRST client-visible success frame the + classified request produces - a request-scoped notification (progress and + the `subscriptions/listen` ack ride those) or its result - immediately + before that frame is written, with no checkpoint in between. One + definition for every method: a plain request locks at its response + exactly as before, and a streaming request (`subscriptions/listen`, whose + response arrives only at close) locks at its ack, so a pipelined + `initialize` behind the ack can never hijack a live stream back to + legacy. For the inline methods (`initialize`, `server/discover`) the + dispatch 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 before any frame + went out does not lock either. + + Classified-modern requests are also governed by the 2026 cancellation + rule for their whole lifetime: after an inbound `notifications/cancelled` + the server sends nothing further for that request - no result, no error + (the transport spec's MUST NOT). Legacy-era requests keep the + byte-identical "Request cancelled" answer released 2025 clients expect. """ dispatcher: JSONRPCDispatcher[TransportContext] = JSONRPCDispatcher( read_stream, @@ -601,28 +620,37 @@ async def serve_dual_era_loop( 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 + # The one definition of "this request may lock the era": it produced # 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. + # cancel means the client already abandoned this request and will + # never read the frame as a success: no lock. return era == "unlocked" and not dctx.cancel_requested.is_set() + def commit_modern(dctx: DispatchContext[TransportContext], version: str) -> None: + # The success-frame observer for classified-modern requests: the + # dispatcher invokes it immediately before each non-error frame for + # the request (ack/event notifications, progress, or the result), + # with no checkpoint before the write - so by the time the client can + # read any modern output, the era is already locked and a pipelined + # `initialize` is rejected instead of hijacking the connection. + nonlocal era, modern_version + if era_settles(dctx): + era, modern_version = "modern", version + async def serve_modern( dctx: DispatchContext[TransportContext], method: str, params: Mapping[str, Any] | None ) -> dict[str, Any]: - nonlocal era, modern_version 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 - ) + # Classification succeeded: 2026 wire semantics govern this request's + # lifecycle from here. Both facts attach synchronously - no checkpoint + # since the dispatcher entered the handler scope - so a peer cancel + # can never interleave and observe the legacy defaults. + suppress_cancel_answer(dctx) + observe_success_frames(dctx, partial(commit_modern, dctx, route.protocol_version)) connection = Connection.from_envelope( route.protocol_version, route.client_info, @@ -630,7 +658,7 @@ async def serve_modern( outbound=standalone_outbound, ) try: - result = await serve_one( + return await serve_one( server, _NoServerRequestsDispatchContext(dctx), method, @@ -647,9 +675,6 @@ async def serve_modern( 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 diff --git a/src/mcp/server/subscriptions.py b/src/mcp/server/subscriptions.py index 6b0b3d49b5..718d26eaf9 100644 --- a/src/mcp/server/subscriptions.py +++ b/src/mcp/server/subscriptions.py @@ -164,8 +164,9 @@ 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). + Requires a transport that can stream a request's response: streamable + HTTP's SSE mode, or a duplex stream pair (stdio) where the listen request + simply stays pending while notifications ride the pipe. `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 29d3f07fa6..73b10935d0 100644 --- a/tests/server/test_runner.py +++ b/tests/server/test_runner.py @@ -16,6 +16,7 @@ import anyio import anyio.abc import pytest +from anyio.streams.memory import MemoryObjectReceiveStream, MemoryObjectSendStream from mcp_types import ( CLIENT_CAPABILITIES_META_KEY, CLIENT_INFO_META_KEY, @@ -64,11 +65,13 @@ serve_one, ) from mcp.server.session import ServerSession -from mcp.shared.dispatcher import CallOptions +from mcp.server.subscriptions import InMemorySubscriptionBus, ListenHandler +from mcp.shared.dispatcher import CallOptions, DispatchContext, OnNotify from mcp.shared.exceptions import MCPError, NoBackChannelError from mcp.shared.jsonrpc_dispatcher import JSONRPCDispatcher from mcp.shared.message import MessageMetadata, SessionMessage from mcp.shared.peer import dump_params +from mcp.shared.subscriptions import SUBSCRIPTION_ID_META_KEY, PromptsListChanged, ToolsListChanged from mcp.shared.transport_context import TransportContext from ..shared.conftest import jsonrpc_pair @@ -1254,12 +1257,15 @@ def _modern_params(version: str = LATEST_MODERN_VERSION, **params: Any) -> dict[ @asynccontextmanager -async def dual_era_client(server: SrvT) -> AsyncIterator[tuple[JSONRPCDispatcher[TransportContext], Recorder]]: +async def dual_era_client( + server: SrvT, *, client_on_notify: OnNotify | None = None +) -> AsyncIterator[tuple[JSONRPCDispatcher[TransportContext], Recorder]]: """Yield `(client, recorder)` speaking raw frames to a `serve_dual_era_loop` server. The driver owns its dispatcher and connection, so unlike `connected_runner` the harness hands it bare streams and performs no handshake: each test - drives the era lock itself. + drives the era lock itself. `client_on_notify` replaces the recorder's + notification handler for tests that assert on server-pushed frames. """ c2s_send, c2s_recv = anyio.create_memory_object_stream[SessionMessage | Exception](32) s2c_send, s2c_recv = anyio.create_memory_object_stream[SessionMessage | Exception](32) @@ -1272,7 +1278,7 @@ def builder(_meta: object) -> TransportContext: c_req, c_notify = echo_handlers(recorder) body_exc: BaseException | None = None async with anyio.create_task_group() as tg: - await tg.start(client.run, c_req, c_notify) + await tg.start(client.run, c_req, client_on_notify if client_on_notify is not None else c_notify) tg.start_soon(partial(serve_dual_era_loop, server, c2s_recv, s2c_send, lifespan_state=_LIFESPAN)) try: with anyio.fail_after(5): @@ -1417,18 +1423,302 @@ 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 def test_dual_era_loop_listen_without_a_handler_is_kernel_method_not_found_and_never_locks(server: SrvT): + """`subscriptions/listen` dispatches through the kernel like every other + modern method: with no listen handler registered it gets the kernel's own + METHOD_NOT_FOUND shape (no transport-level carve-out), 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()) + await client.send_raw_request( + "subscriptions/listen", _modern_params(notifications={"toolsListChanged": True}) + ) + init = await client.send_raw_request("initialize", _initialize_params()) + assert init["protocolVersion"] == LATEST_HANDSHAKE_VERSION + assert exc_info.value.error == ErrorData( + code=METHOD_NOT_FOUND, message="Method not found", data="subscriptions/listen" + ) + + +ACK_METHOD = "notifications/subscriptions/acknowledged" + + +def _modern_listen_params() -> dict[str, Any]: + return _modern_params(notifications={"toolsListChanged": True}) + + +def _note_collector() -> tuple[ + OnNotify, + MemoryObjectSendStream[tuple[str, dict[str, Any]]], + MemoryObjectReceiveStream[tuple[str, dict[str, Any]]], +]: + """An `OnNotify` that forwards each notification into a stream the test can await. + + Returns the handler plus both stream ends; tests own closing them (wrap + the body in `with send, recv:`). + """ + send, recv = anyio.create_memory_object_stream[tuple[str, dict[str, Any]]](32) + + async def on_notify(dctx: DispatchContext[TransportContext], method: str, params: Mapping[str, Any] | None) -> None: + await send.send((method, dict(params or {}))) + + return on_notify, send, recv + + +@pytest.mark.anyio +async def test_dual_era_loop_serves_listen_and_locks_the_era_at_the_ack(): + """The full listen lifecycle over the stream loop: the request stays + pending, the ack (stamped with the request id verbatim, honored subset + echoed) is the first frame on the subscription, events within the filter + are delivered stamped, kinds the client did not request never leave the + server, and `ListenHandler.close()` finally resolves the request with the + stamped SubscriptionsListenResult. The era lock commits at the ACK - not + at the (much later) response - so an initialize pipelined behind a live + stream is rejected with -32022 instead of hijacking the connection.""" + bus = InMemorySubscriptionBus() + handler = ListenHandler(bus) + srv: SrvT = Server(name="listen-server", version="0.0.1", on_subscriptions_listen=handler) + on_notify, notes_send, notes = _note_collector() + result_box: list[dict[str, Any]] = [] + try: + async with dual_era_client(srv, client_on_notify=on_notify) as (client, _): + async with anyio.create_task_group() as tg: + + async def open_listen() -> None: + result_box.append( + await client.send_raw_request( + "subscriptions/listen", _modern_listen_params(), {"request_id": "sub-1"} + ) + ) + + tg.start_soon(open_listen) + method, params = await notes.receive() + assert method == ACK_METHOD + assert params["notifications"] == {"toolsListChanged": True} + assert params["_meta"][SUBSCRIPTION_ID_META_KEY] == "sub-1" + # The ack committed the era lock while the listen request is + # still pending: a pipelined initialize cannot flip the + # connection legacy under the live stream. + with pytest.raises(MCPError) as init_exc: + await client.send_raw_request("initialize", _initialize_params()) + assert init_exc.value.error.code == UNSUPPORTED_PROTOCOL_VERSION + # Deliveries preserve publish order, so if the un-requested + # prompts event leaked it would arrive before the tools event + # asserted here. + await bus.publish(PromptsListChanged()) + await bus.publish(ToolsListChanged()) + method, params = await notes.receive() + assert method == "notifications/tools/list_changed" + assert params["_meta"][SUBSCRIPTION_ID_META_KEY] == "sub-1" + handler.close() + finally: + notes_send.close() + notes.close() + assert result_box[0]["resultType"] == "complete" + assert result_box[0]["_meta"][SUBSCRIPTION_ID_META_KEY] == "sub-1" + + +@pytest.mark.anyio +async def test_dual_era_loop_listen_cancel_is_total_silence_and_frees_the_subscription(): + """Client cancel of a live listen: the server stops delivery, frees the + subscription slot, and sends NOTHING further for that id - no result, no + error - while the connection stays healthy for other traffic.""" + bus = InMemorySubscriptionBus() + handler = ListenHandler(bus, max_subscriptions=1) + + async def list_tools(ctx: Ctx, params: PaginatedRequestParams | None) -> ListToolsResult: + return ListToolsResult(tools=[Tool(name="t", input_schema={"type": "object"})]) + + srv: SrvT = Server(name="listen-server", version="0.0.1", on_subscriptions_listen=handler, on_list_tools=list_tools) + on_notify, notes_send, notes = _note_collector() + first_outcome: list[Any] = [] + second_outcome: list[dict[str, Any]] = [] + try: + async with dual_era_client(srv, client_on_notify=on_notify) as (client, _): + abandoned = anyio.CancelScope() + async with anyio.create_task_group() as tg: + + async def open_first() -> None: + # A (buggy) late error frame for sub-1 would raise MCPError + # here and fail the test through the task group; a late + # result frame lands in first_outcome. Silence leaves the + # call parked until `abandoned` is cancelled. + with abandoned: + first_outcome.append( + await client.send_raw_request( + "subscriptions/listen", _modern_listen_params(), {"request_id": "sub-1"} + ) + ) + + tg.start_soon(open_first) + method, params = await notes.receive() + assert method == ACK_METHOD + assert params["_meta"][SUBSCRIPTION_ID_META_KEY] == "sub-1" + await client.notify("notifications/cancelled", {"requestId": "sub-1"}) + # Ordering marker: this round-trip proves the server consumed + # the cancel, and shows the connection stays healthy. + result = await client.send_raw_request("tools/list", _modern_params()) + assert result["tools"][0]["name"] == "t" + # Once the cancelled handler task has fully unwound, its + # max_subscriptions=1 slot must be free for a new listen. + await anyio.wait_all_tasks_blocked() + + async def open_second() -> None: + second_outcome.append( + await client.send_raw_request( + "subscriptions/listen", _modern_listen_params(), {"request_id": "sub-2"} + ) + ) + + tg.start_soon(open_second) + method, params = await notes.receive() + assert method == ACK_METHOD + assert params["_meta"][SUBSCRIPTION_ID_META_KEY] == "sub-2" + handler.close() + # Give any (buggy) late frame for sub-1 every chance to land + # before asserting silence: sub-2's close resolution below is + # ordered after anything the server wrote earlier. + await anyio.wait_all_tasks_blocked() + assert first_outcome == [] + abandoned.cancel() + finally: + notes_send.close() + notes.close() + assert second_outcome[0]["_meta"][SUBSCRIPTION_ID_META_KEY] == "sub-2" + assert first_outcome == [] + + +@pytest.mark.anyio +async def test_dual_era_loop_modern_peer_cancel_is_silent_and_never_locks_without_a_frame(): + """The 2026 cancellation rule is era-scoped, not listen-specific: a + peer-cancelled modern tools/list gets NO late answer (the legacy + "Request cancelled" frame is handshake-era only), and a request cancelled + before it produced any frame never locks the era.""" + entered = anyio.Event() + + async def parked(ctx: Ctx, params: PaginatedRequestParams | None) -> ListToolsResult: + entered.set() + await anyio.sleep_forever() + raise NotImplementedError + + srv: SrvT = Server(name="parked-server", version="0.0.1", on_list_tools=parked) + outcome: list[Any] = [] + async with dual_era_client(srv) as (client, _): + abandoned = anyio.CancelScope() + async with anyio.create_task_group() as tg: + + async def call() -> None: + # A (buggy) late error frame would raise MCPError here and + # fail the test through the task group; a late result lands + # in `outcome`. Silence leaves the call parked until + # `abandoned` is cancelled. + with abandoned: + outcome.append(await client.send_raw_request("tools/list", _modern_params(), {"request_id": 41})) + + tg.start_soon(call) + await entered.wait() + await client.notify("notifications/cancelled", {"requestId": 41}) + # Ordering marker: the error round-trip proves the server consumed + # the cancel before we assert on the cancelled id's silence. + with pytest.raises(MCPError): + await client.send_raw_request("probe/marker", None) + await anyio.wait_all_tasks_blocked() + assert outcome == [] + # No frame ever went out for the cancelled request, so the era + # never locked: the legacy handshake is still available. + init = await client.send_raw_request("initialize", _initialize_params()) + assert init["protocolVersion"] == LATEST_HANDSHAKE_VERSION + abandoned.cancel() + assert outcome == [] + + +@pytest.mark.anyio +async def test_dual_era_loop_legacy_cancel_answer_is_byte_identical(): + """The 2026 silence rule is scoped to classified-modern requests: on a + handshake-era connection a peer-cancelled request still gets the exact + legacy answer released 2025 clients unblock on.""" + entered = anyio.Event() + + async def parked(ctx: Ctx, params: PaginatedRequestParams | None) -> ListToolsResult: + entered.set() + await anyio.sleep_forever() + raise NotImplementedError + + srv: SrvT = Server(name="parked-server", version="0.0.1", on_list_tools=parked) + errors: list[ErrorData] = [] + async with dual_era_client(srv) as (client, _): + await client.send_raw_request("initialize", _initialize_params()) + async with anyio.create_task_group() as tg: + + async def call() -> None: + with pytest.raises(MCPError) as exc_info: + await client.send_raw_request("tools/list", None, {"request_id": 9}) + errors.append(exc_info.value.error) + + tg.start_soon(call) + await entered.wait() + await client.notify("notifications/cancelled", {"requestId": 9}) + assert errors == [ErrorData(code=0, message="Request cancelled")] + + +@pytest.mark.anyio +async def test_dual_era_loop_progress_frame_locks_modern_even_when_the_request_later_fails(server: SrvT): + """The era lock rides the first client-visible frame, not the response: a + modern request that reports progress and then fails has already shown the + client modern output, so the era is locked despite the failure (a client + that stamped a valid envelope committed to modern).""" + + async def progress_then_fail(ctx: Ctx, params: RequestParams | None) -> dict[str, Any]: + await ctx.session.report_progress(0.5) + raise MCPError(code=INTERNAL_ERROR, message="late failure") + + server.add_request_handler("x/progress", RequestParams, progress_then_fail) + params = _modern_params() + params["_meta"]["progressToken"] = "tok-1" + async with dual_era_client(server) as (client, recorder): + with pytest.raises(MCPError) as exc_info: + await client.send_raw_request("x/progress", params) + assert exc_info.value.error.code == INTERNAL_ERROR + await recorder.notified.wait() + assert recorder.notifications[0][0] == "notifications/progress" + with pytest.raises(MCPError) as init_exc: + await client.send_raw_request("initialize", _initialize_params()) + assert init_exc.value.error.code == UNSUPPORTED_PROTOCOL_VERSION + + +@pytest.mark.anyio +async def test_dual_era_loop_progress_without_a_token_is_dropped_and_never_locks(server: SrvT): + """`report_progress` with no client progressToken writes nothing, and a + frame that never reached the wire must not lock: only client-visible + output commits the era, so the failed request leaves the handshake open.""" + + async def progress_then_fail(ctx: Ctx, params: RequestParams | None) -> dict[str, Any]: + await ctx.session.report_progress(0.5) + raise MCPError(code=INTERNAL_ERROR, message="late failure") + + server.add_request_handler("x/progress", RequestParams, progress_then_fail) + async with dual_era_client(server) as (client, _): + with pytest.raises(MCPError): + await client.send_raw_request("x/progress", _modern_params()) + init = await client.send_raw_request("initialize", _initialize_params()) + assert init["protocolVersion"] == LATEST_HANDSHAKE_VERSION + + +@pytest.mark.anyio +async def test_dual_era_loop_capacity_rejected_listen_never_locks(): + """A listen rejected before its ack (subscription limit) produced no + client-visible frame: like every failed modern request it must not lock + the era, so the legacy handshake stays available.""" + bus = InMemorySubscriptionBus() + handler = ListenHandler(bus, max_subscriptions=0) + srv: SrvT = Server(name="full-server", version="0.0.1", on_subscriptions_listen=handler) + async with dual_era_client(srv) as (client, _): + with pytest.raises(MCPError) as exc_info: + await client.send_raw_request("subscriptions/listen", _modern_listen_params(), {"request_id": "sub-1"}) + assert exc_info.value.error.code == INTERNAL_ERROR 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 @pytest.mark.anyio @@ -1579,11 +1869,12 @@ async def modern_call() -> None: @pytest.mark.anyio -async def test_dual_era_loop_modern_success_cancelled_away_at_the_response_write_never_locks(): +async def test_dual_era_loop_modern_success_cancelled_away_at_the_response_write_is_silent_and_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.""" + client already abandoned the request: at 2026 the dispatcher sends + NOTHING for it - not the computed result, not the legacy "Request + cancelled" frame - and the era must not lock, so the legacy handshake + stays available.""" entered = anyio.Event() release = anyio.Event() @@ -1597,28 +1888,35 @@ async def list_tools(ctx: Ctx, params: PaginatedRequestParams | None) -> ListToo return ListToolsResult(tools=[]) parked = Server(name="parked-server", version="0.0.1", on_list_tools=list_tools) + outcome: list[Any] = [] async with dual_era_client(parked) as (client, _): - failures: list[MCPError] = [] + abandoned = anyio.CancelScope() + async with anyio.create_task_group() as tg: - 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 def modern_call() -> None: + # A (buggy) late "Request cancelled" frame would raise + # MCPError here and fail the test through the task group; a + # late result lands in `outcome`. Silence leaves the call + # parked until `abandoned` is cancelled. + with abandoned: + outcome.append(await client.send_raw_request("tools/list", _modern_params(), {"request_id": 51})) - 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}) + await client.notify("notifications/cancelled", {"requestId": 51}) # 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) 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 + await anyio.wait_all_tasks_blocked() + # Total silence for the cancelled id: the handler's success was + # computed but never became client-visible - and never locked. + assert outcome == [] + init = await client.send_raw_request("initialize", _initialize_params()) + assert init["protocolVersion"] == LATEST_HANDSHAKE_VERSION + abandoned.cancel() + assert outcome == [] @pytest.mark.anyio From 7c6aba773ad03acb749252c3b28dc8e1a8a540c1 Mon Sep 17 00:00:00 2001 From: Max Isbey <224885523+maxisbey@users.noreply.github.com> Date: Thu, 16 Jul 2026 22:23:17 +0000 Subject: [PATCH 3/7] Document and pin the accepted orphaned-modern-lock corner A peer cancel that lands at the first success frame's own transport-write checkpoint - after the era lock committed, before the frame delivered - leaves a dual-era connection locked modern with ZERO frames on the wire. era_settles already re-checks cancel_requested at observer fire time, so this one-checkpoint window is the entire residue, and it exists only for mid-handler frames (the listen ack, progress): a plain request's response cannot be cancelled away because its in-flight entry is removed, with no checkpoint in between, before the response write starts. This is a deliberate, documented deviation from the no-frame-no-lock doctrine: closing it fully requires committing the lock only after the write is known delivered, which reopens the worse pipelined-initialize hijack the at-first-frame lock exists to prevent. The orphaned state is self-consistent and client-recoverable - only a validly-classified modern envelope reaches this path, a follow-up initialize is answered with -32022 naming the modern versions (released auto-negotiating clients treat that as modern evidence and re-probe), and the connection keeps serving modern traffic. Document the corner on serve_dual_era_loop and commit_modern, and pin it with raw-frame adjacency tests: the orphan itself (locked, silent, recoverable, slot freed), the ack beating an adjacent pipelined initialize, and the plain-request path that cannot orphan. --- src/mcp/server/runner.py | 42 ++++++-- tests/server/test_runner.py | 190 ++++++++++++++++++++++++++++++++++++ 2 files changed, 224 insertions(+), 8 deletions(-) diff --git a/src/mcp/server/runner.py b/src/mcp/server/runner.py index f03b39e408..826eebc085 100644 --- a/src/mcp/server/runner.py +++ b/src/mcp/server/runner.py @@ -595,8 +595,33 @@ async def serve_dual_era_loop( 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 before any frame - went out does not lock either. + does not move; and a peer cancel observed before the first frame's write + begins prevents the lock (the commit re-checks `cancel_requested` at fire + time). + + One residual corner is accepted, deliberately: the lock commits + immediately before the frame's transport write, and that write's own + checkpoint is the first point a pending peer cancel can land - so a + cancel arriving DURING the write cancels the frame away after the lock + already committed, leaving the connection locked modern with ZERO frames + delivered. Only mid-handler frames (the listen ack, progress) have this + window; a plain request's response cannot be cancelled away (its + in-flight entry is removed before the response write starts, with no + checkpoint in between, so a peer cancel no longer reaches it). The + converse exists too: a handler that shields its own send past a pending + cancel delivers a frame WITHOUT the lock - `era_settles` declined at fire + time - but that frame is output the handler forced after the cancel said + stop, not a state this loop produces. The + orphaned lock is self-consistent and client-recoverable: only a + validly-classified modern envelope reaches this path, so the peer already + committed to modern; a follow-up `initialize` is answered with + UNSUPPORTED_PROTOCOL_VERSION (-32022) naming the modern versions, which + released auto-negotiating clients treat as modern evidence and re-probe; + and the connection keeps serving modern traffic. Closing the corner would + require committing the lock only after the write is known delivered, + which reopens the worse race this design exists to prevent (an + `initialize` pipelined behind an ack the client already read, flipping a + live stream legacy). Classified-modern requests are also governed by the 2026 cancellation rule for their whole lifetime: after an inbound `notifications/cancelled` @@ -629,12 +654,13 @@ def era_settles(dctx: DispatchContext[TransportContext]) -> bool: return era == "unlocked" and not dctx.cancel_requested.is_set() def commit_modern(dctx: DispatchContext[TransportContext], version: str) -> None: - # The success-frame observer for classified-modern requests: the - # dispatcher invokes it immediately before each non-error frame for - # the request (ack/event notifications, progress, or the result), - # with no checkpoint before the write - so by the time the client can - # read any modern output, the era is already locked and a pipelined - # `initialize` is rejected instead of hijacking the connection. + # The success-frame observer for classified-modern requests: invoked + # immediately before each non-error frame's write, with no checkpoint + # in between - by the time the client can read any modern output, the + # era is locked and a pipelined `initialize` is rejected instead of + # hijacking the connection. `era_settles` re-checks `cancel_requested` + # at fire time; the cancel-during-the-write residue is the accepted + # corner documented on `serve_dual_era_loop`. nonlocal era, modern_version if era_settles(dctx): era, modern_version = "modern", version diff --git a/tests/server/test_runner.py b/tests/server/test_runner.py index 73b10935d0..4b1de10002 100644 --- a/tests/server/test_runner.py +++ b/tests/server/test_runner.py @@ -31,14 +31,19 @@ ErrorData, Implementation, InitializeRequestParams, + JSONRPCError, + JSONRPCNotification, JSONRPCRequest, + JSONRPCResponse, ListToolsResult, NotificationParams, PaginatedRequestParams, ProgressNotificationParams, + RequestId, RequestParams, SetLevelRequestParams, Tool, + UnsupportedProtocolVersionErrorData, ) from mcp_types.version import ( LATEST_HANDSHAKE_VERSION, @@ -1466,6 +1471,57 @@ async def on_notify(dctx: DispatchContext[TransportContext], method: str, params return on_notify, send, recv +def _raw_req(rid: RequestId, method: str, params: dict[str, Any] | None = None) -> SessionMessage: + return SessionMessage(message=JSONRPCRequest(jsonrpc="2.0", id=rid, method=method, params=params)) + + +def _raw_note(method: str, params: dict[str, Any] | None = None) -> SessionMessage: + return SessionMessage(message=JSONRPCNotification(jsonrpc="2.0", method=method, params=params)) + + +async def _expect_ack(s2c: MemoryObjectReceiveStream[SessionMessage], subscription_id: RequestId) -> None: + """Receive the next frame and assert it is the ack stamped with `subscription_id`.""" + ack = (await s2c.receive()).message + assert isinstance(ack, JSONRPCNotification) + assert ack.method == ACK_METHOD + assert ack.params is not None + assert ack.params["_meta"][SUBSCRIPTION_ID_META_KEY] == subscription_id + + +@asynccontextmanager +async def raw_dual_era_loop( + server: SrvT, +) -> AsyncIterator[ + tuple[MemoryObjectSendStream[SessionMessage | Exception], MemoryObjectReceiveStream[SessionMessage]] +]: + """Yield raw `(c2s_send, s2c_recv)` streams around a live `serve_dual_era_loop`. + + For frame-adjacency tests: `send_nowait` places frames in the loop's + buffer back-to-back, so races are driven at the wire with no client + dispatcher reordering between them. Assert absence with + `anyio.wait_all_tasks_blocked()` + `receive_nowait` (every frame the + server will ever write for the consumed input is buffered once all tasks + are parked), never with timed drains. + """ + c2s_send, c2s_recv = anyio.create_memory_object_stream[SessionMessage | Exception](32) + s2c_send, s2c_recv = anyio.create_memory_object_stream[SessionMessage](32) + body_exc: BaseException | None = None + try: + async with anyio.create_task_group() as tg: + tg.start_soon(partial(serve_dual_era_loop, server, c2s_recv, s2c_send, lifespan_state=_LIFESPAN)) + try: + with anyio.fail_after(5): + yield c2s_send, s2c_recv + except BaseException as e: + body_exc = e + tg.cancel_scope.cancel() + if body_exc is not None: + raise body_exc + finally: + for stream in (c2s_send, c2s_recv, s2c_send, s2c_recv): + stream.close() + + @pytest.mark.anyio async def test_dual_era_loop_serves_listen_and_locks_the_era_at_the_ack(): """The full listen lifecycle over the stream loop: the request stays @@ -1919,6 +1975,132 @@ async def modern_call() -> None: assert outcome == [] +@pytest.mark.anyio +async def test_dual_era_loop_cancel_racing_the_listen_ack_write_pins_the_accepted_orphaned_modern_lock(): + """ACCEPTED DOCTRINE DEVIATION, pinned deliberately: a peer cancel landing + at the ack write's own checkpoint - after `commit_modern` ran, before the + frame delivered - leaves the era locked modern with ZERO frames on the + wire. `era_settles` re-checks `cancel_requested` at fire time, so this is + the only remaining window; closing it entirely needs a post-write commit, + which reopens the pipelined-initialize hijack the at-first-frame lock + exists to prevent (see the residual-corner paragraph on + `serve_dual_era_loop`). What this test pins is that the orphaned state is + self-consistent and client-recoverable: total silence for the cancelled + listen, its subscription slot freed, a follow-up initialize answered with + -32022 naming the modern versions (released auto-negotiating clients + treat that as modern evidence and re-probe), and the connection still + serving modern traffic. Only a validly-classified modern envelope reaches + this path, so the peer already committed to modern and cannot be + bricked. + + Steps: + 1. listen + cancel adjacent in the buffer -> zero frames delivered. + 2. initialize -> -32022 with the typed payload (era IS locked). + 3. fresh listen acks (slot freed) and resolves on graceful close. + """ + bus = InMemorySubscriptionBus() + handler = ListenHandler(bus, max_subscriptions=1) + srv: SrvT = Server(name="listen-server", version="0.0.1", on_subscriptions_listen=handler) + async with raw_dual_era_loop(srv) as (c2s, s2c): + # Adjacent in the buffer before the server reads either: the loop + # spawns the listen handler, the handler commits the modern lock + # immediately before the ack write, the write's checkpoint yields to + # the read loop, and the read loop's cancel then cancels the ack away. + c2s.send_nowait(_raw_req("L", "subscriptions/listen", _modern_listen_params())) + c2s.send_nowait(_raw_note("notifications/cancelled", {"requestId": "L"})) + await anyio.wait_all_tasks_blocked() + # Zero frames delivered: no ack, and no response/error for the + # cancelled listen (the 2026 silence rule). + with pytest.raises(anyio.WouldBlock): + s2c.receive_nowait() + # The era is locked modern despite the empty wire: initialize gets the + # typed -32022 rejection naming what the connection serves. + c2s.send_nowait(_raw_req(2, "initialize", _initialize_params())) + init_answer = (await s2c.receive()).message + assert isinstance(init_answer, JSONRPCError) + assert init_answer.id == 2 + assert init_answer.error.code == UNSUPPORTED_PROTOCOL_VERSION + assert init_answer.error.data == UnsupportedProtocolVersionErrorData( + supported=list(MODERN_PROTOCOL_VERSIONS), requested=LATEST_HANDSHAKE_VERSION + ).model_dump(mode="json") + # Recoverable: the cancelled listen's max_subscriptions=1 slot was + # freed, and the connection serves modern traffic - a new listen acks + # and resolves gracefully. + c2s.send_nowait(_raw_req("L2", "subscriptions/listen", _modern_listen_params())) + await _expect_ack(s2c, "L2") + handler.close() + result = (await s2c.receive()).message + assert isinstance(result, JSONRPCResponse) + assert result.id == "L2" + + +@pytest.mark.anyio +async def test_dual_era_loop_initialize_pipelined_adjacent_behind_a_listen_cannot_hijack_the_stream(): + """listen + initialize adjacent in the loop's buffer, before the ack write + could complete: the modern lock commits immediately before the ack write + starts, and initialize runs inline only after the read loop resumes - so + modern wins, the initialize is rejected with -32022, the rejection is + sticky, and the listen stream stays live underneath it.""" + bus = InMemorySubscriptionBus() + handler = ListenHandler(bus) + srv: SrvT = Server(name="listen-server", version="0.0.1", on_subscriptions_listen=handler) + async with raw_dual_era_loop(srv) as (c2s, s2c): + c2s.send_nowait(_raw_req("L", "subscriptions/listen", _modern_listen_params())) + c2s.send_nowait(_raw_req(2, "initialize", _initialize_params())) + # Exactly two frames, in either write order: the ack for the live + # listen and the -32022 for the hijack attempt. Nothing for id "L". + frames = [(await s2c.receive()).message, (await s2c.receive()).message] + (ack,) = [f for f in frames if isinstance(f, JSONRPCNotification)] + (init_answer,) = [f for f in frames if isinstance(f, JSONRPCError)] + assert ack.method == ACK_METHOD + assert ack.params is not None + assert ack.params["_meta"][SUBSCRIPTION_ID_META_KEY] == "L" + assert init_answer.id == 2 + assert init_answer.error.code == UNSUPPORTED_PROTOCOL_VERSION + # Sticky: a retried initialize is rejected identically. + c2s.send_nowait(_raw_req(3, "initialize", _initialize_params())) + follow = (await s2c.receive()).message + assert isinstance(follow, JSONRPCError) + assert follow.id == 3 + assert follow.error.code == UNSUPPORTED_PROTOCOL_VERSION + # The stream the initialize tried to hijack is genuinely live. + await bus.publish(ToolsListChanged()) + event = (await s2c.receive()).message + assert isinstance(event, JSONRPCNotification) + assert event.method == "notifications/tools/list_changed" + assert event.params is not None + assert event.params["_meta"][SUBSCRIPTION_ID_META_KEY] == "L" + handler.close() + result = (await s2c.receive()).message + assert isinstance(result, JSONRPCResponse) + assert result.id == "L" + + +@pytest.mark.anyio +async def test_dual_era_loop_adjacent_cancel_behind_a_fast_modern_request_cannot_orphan_the_lock(): + """The orphaned-lock corner is confined to mid-handler frames (ack, + progress): a plain request's in-flight entry is removed before its + response write starts, with no checkpoint in between, so an adjacent + cancel consumed during the write no longer reaches the request - the + response is delivered and the era locks WITH a frame on the wire.""" + + async def list_tools(ctx: Ctx, params: PaginatedRequestParams | None) -> ListToolsResult: + return ListToolsResult(tools=[Tool(name="t", input_schema={"type": "object"})]) + + srv: SrvT = Server(name="fast-server", version="0.0.1", on_list_tools=list_tools) + async with raw_dual_era_loop(srv) as (c2s, s2c): + c2s.send_nowait(_raw_req(5, "tools/list", _modern_params())) + c2s.send_nowait(_raw_note("notifications/cancelled", {"requestId": 5})) + result = (await s2c.receive()).message + assert isinstance(result, JSONRPCResponse) + assert result.id == 5 + c2s.send_nowait(_raw_req(6, "initialize", _initialize_params())) + answer = (await s2c.receive()).message + assert isinstance(answer, JSONRPCError) + assert answer.id == 6 + assert answer.error.code == UNSUPPORTED_PROTOCOL_VERSION + + @pytest.mark.anyio async def test_dual_era_loop_maps_unmapped_handler_exceptions_like_the_modern_http_entry(): """An unmapped handler exception on a modern request surfaces as the @@ -2071,3 +2253,11 @@ async def test_dual_era_client_propagates_body_exception_unwrapped(server: SrvT) with pytest.raises(RuntimeError, match="boom"): async with dual_era_client(server): raise RuntimeError("boom") + + +@pytest.mark.anyio +async def test_raw_dual_era_loop_propagates_body_exception_unwrapped(server: SrvT): + """The raw-frame harness re-raises body exceptions as-is, not as `ExceptionGroup`.""" + with pytest.raises(RuntimeError, match="boom"): + async with raw_dual_era_loop(server): + raise RuntimeError("boom") From 34d4f7407c911b2933d2eb5915c83540fc4b17c7 Mon Sep 17 00:00:00 2001 From: Max Isbey <224885523+maxisbey@users.noreply.github.com> Date: Thu, 16 Jul 2026 22:24:42 +0000 Subject: [PATCH 4/7] Honor the cancel-silence commitment in the shutdown drain The dispatcher's shutdown arm wrote the CONNECTION_CLOSED drain answer for every in-flight request it caught unwinding, including one whose peer had already cancelled it and whose loop layer had committed it to 2026 cancel semantics - where no frame of any kind may follow notifications/cancelled. A silence-committed peer-cancelled request caught at loop shutdown now stays silent. The rule lives in one predicate on the dispatch context, must_stay_silent, consulted by both frame-suppressing sites (_answer_error and the shutdown drain) so it cannot drift between them; the interrupt arm remains its complementary write side. Requests the peer did NOT cancel keep the drain answer regardless of the commitment - the EOF-drain doctrine is unchanged. --- src/mcp/shared/jsonrpc_dispatcher.py | 26 +++++++++-- tests/shared/test_jsonrpc_dispatcher.py | 58 +++++++++++++++++++++++++ 2 files changed, 80 insertions(+), 4 deletions(-) diff --git a/src/mcp/shared/jsonrpc_dispatcher.py b/src/mcp/shared/jsonrpc_dispatcher.py index 1f3b29479f..76d78c2e06 100644 --- a/src/mcp/shared/jsonrpc_dispatcher.py +++ b/src/mcp/shared/jsonrpc_dispatcher.py @@ -179,6 +179,20 @@ def request_id(self) -> RequestId | None: def can_send_request(self) -> bool: return self.transport.can_send_request and not self._closed + @property + def must_stay_silent(self) -> bool: + """The peer cancelled this silence-committed request: nothing may be written for it. + + Consulted by both frame-suppressing sites - `_answer_error` and the + shutdown drain. The interrupt arm expresses the same fact from the + write side: within `scope.cancelled_caught`, `cancel_requested` is + definitionally set, so its `cancel_answer is not None` check is + exactly `not must_stay_silent`. Once true, result, error, and drain + frames for this id are all forbidden - the transport spec's MUST NOT + after `notifications/cancelled`. + """ + return self.cancel_answer is None and self.cancel_requested.is_set() + async def notify(self, method: str, params: Mapping[str, Any] | None, opts: CallOptions | None = None) -> None: if self._closed: logger.debug("dropped %s: dispatch context closed", method) @@ -796,9 +810,13 @@ async def _handle_request( except anyio.get_cancelled_exc_class(): # Shutdown: answer the request so the peer isn't left waiting - unless # an answer write already started (it may have reached the transport; - # prefer possibly-zero answers over possibly-two). The shielded helper - # is needed because bare awaits re-raise here. - if not answer_write_started: + # prefer possibly-zero answers over possibly-two), or the peer already + # cancelled a silence-committed request (the 2026 rule forbids ANY + # frame after the cancel, the shutdown drain included; requests the + # peer did NOT cancel keep their drain answer regardless of the + # commitment). The shielded helper is needed because bare awaits + # re-raise here. + if not answer_write_started and not dctx.must_stay_silent: await self._final_write( partial(self._write_error, req.id, ErrorData(code=CONNECTION_CLOSED, message="Connection closed")), shield=True, @@ -848,7 +866,7 @@ async def _answer_error( failure was computed before the cancellation interrupt could land (a synchronous raise after the cancel arrived, or a shielded handler). """ - if dctx.cancel_answer is None and dctx.cancel_requested.is_set(): + if dctx.must_stay_silent: logger.debug("dropped error for %r: peer cancelled a silence-committed request", request_id) return await self._write_error(request_id, error) diff --git a/tests/shared/test_jsonrpc_dispatcher.py b/tests/shared/test_jsonrpc_dispatcher.py index 4deb6fccfb..bec38992f2 100644 --- a/tests/shared/test_jsonrpc_dispatcher.py +++ b/tests/shared/test_jsonrpc_dispatcher.py @@ -308,6 +308,64 @@ async def on_request(ctx: DCtx, method: str, params: Mapping[str, Any] | None) - assert await client.send_raw_request("m", None) == {"ok": True} +@pytest.mark.anyio +async def test_shutdown_drain_stays_silent_for_a_peer_cancelled_silence_committed_request(): + """Loop shutdown catching a silence-committed request whose peer already + cancelled writes NOTHING for it: the 2026 rule forbids any frame after + the cancel, the CONNECTION_CLOSED drain answer included. Requests the + peer did NOT cancel keep the drain answer regardless of the commitment + (see test_shutdown_answers_in_flight_request_with_connection_closed).""" + c2s_send, c2s_recv = anyio.create_memory_object_stream[SessionMessage | Exception](32) + s2c_send, s2c_recv = anyio.create_memory_object_stream[SessionMessage | Exception](32) + server: JSONRPCDispatcher[TransportContext] = JSONRPCDispatcher(c2s_recv, s2c_send) + handler_started = anyio.Event() + release_shield = anyio.Event() + + async def on_request(ctx: DCtx, method: str, params: Mapping[str, Any] | None) -> dict[str, Any]: + suppress_cancel_answer(ctx) + handler_started.set() + # Survive the peer-cancel interrupt inside the shield, then hold the + # request in flight until the loop shutdown catches it unwinding. + with anyio.CancelScope(shield=True): + await ctx.cancel_requested.wait() + await release_shield.wait() + await anyio.sleep_forever() + raise NotImplementedError + + async def on_notify(ctx: DCtx, method: str, params: Mapping[str, Any] | None) -> None: + pass + + late: list[SessionMessage | Exception] = [] + try: + async with anyio.create_task_group() as tg: + await tg.start(server.run, on_request, on_notify) + with anyio.fail_after(5): + await c2s_send.send(SessionMessage(message=JSONRPCRequest(jsonrpc="2.0", id=7, method="slow"))) + await handler_started.wait() + await c2s_send.send( + SessionMessage( + message=JSONRPCNotification( + jsonrpc="2.0", method="notifications/cancelled", params={"requestId": 7} + ) + ) + ) + await anyio.wait_all_tasks_blocked() + # The handler observed the cancel inside its shield; shut the + # loop down while the request is still in flight. + tg.cancel_scope.cancel() + release_shield.set() + while True: + try: + late.append(s2c_recv.receive_nowait()) + except (anyio.WouldBlock, anyio.EndOfStream, anyio.ClosedResourceError): + break + finally: + for s in (c2s_send, c2s_recv, s2c_send, s2c_recv): + s.close() + # Total silence for the cancelled id: no drain frame at shutdown. + assert late == [] + + @pytest.mark.anyio async def test_peer_cancel_landing_after_handlers_last_checkpoint_writes_only_the_result(): """A peer cancel that fails to interrupt the handler writes only the result: one answer per From 4932a44d4e5da5ed8edd38d0a0f425cd1d2b1356 Mon Sep 17 00:00:00 2001 From: Max Isbey <224885523+maxisbey@users.noreply.github.com> Date: Thu, 16 Jul 2026 22:25:33 +0000 Subject: [PATCH 5/7] Widen 2026 cancel silence to every request on a modern-locked connection The silence fact attached only after classification succeeded, so a request rejected before classification (envelope-less, malformed envelope) on a modern-locked connection could still answer after a peer cancel: the reject runs on a task spawned after the read loop may have already consumed an adjacent cancel. Unreachable under asyncio's FIFO scheduling - the reject's answer write always starts before the cancel is consumed, which is the legitimate already-answering case - but reachable under trio's randomized scheduling and under any future checkpoint on the reject path, where the frame written would have been the legacy code-0 answer or a post-cancel rejection error. Attach the fact at the top of the modern-locked branch instead, rejections included: a rejection error is a legitimate answer, only the post-cancel frame must be silent. On a still-unlocked connection the rule stays classification-scoped (an unclassifiable request has not proven modern semantics and keeps the legacy answer), initialize has no window at all (inline dispatch parks the read loop), and a legacy straggler cancelled after the lock keeps its legacy answer - scope documented at both attachment sites and on serve_dual_era_loop. With the locked branch owning the connection-scoped fact, serve_modern now attaches its two per-request facts only while the era is unlocked: once locked, the silence fact is already set and the era-lock observer is a permanent no-op, so post-lock requests skip the partial allocation and the per-frame observer invocation. --- src/mcp/server/runner.py | 42 ++++++++++++++++++++++++++++++------- tests/server/test_runner.py | 41 ++++++++++++++++++++++++++++++++++++ 2 files changed, 76 insertions(+), 7 deletions(-) diff --git a/src/mcp/server/runner.py b/src/mcp/server/runner.py index 826eebc085..037ed23b11 100644 --- a/src/mcp/server/runner.py +++ b/src/mcp/server/runner.py @@ -626,8 +626,19 @@ async def serve_dual_era_loop( Classified-modern requests are also governed by the 2026 cancellation rule for their whole lifetime: after an inbound `notifications/cancelled` the server sends nothing further for that request - no result, no error - (the transport spec's MUST NOT). Legacy-era requests keep the - byte-identical "Request cancelled" answer released 2025 clients expect. + (the transport spec's MUST NOT). Once the connection is LOCKED modern the + rule widens from classification-scoped to connection-scoped: every + request - including one rejected before classification succeeds + (envelope-less, malformed envelope) - carries the silence commitment, so + a rejection error computed after the peer's cancel never reaches the wire + (the rejection itself, absent a cancel, is of course still answered). + Legacy-era requests keep the byte-identical "Request cancelled" answer + released 2025 clients expect - including a legacy-classified request + still in flight when the modern lock commits (the straggler carve-out + covers its cancel answer, not just its response) - as does a request that + fails classification on a still-unlocked connection: an unclassifiable + request has not proven modern semantics, so the legacy answer stands + there. """ dispatcher: JSONRPCDispatcher[TransportContext] = JSONRPCDispatcher( read_stream, @@ -672,11 +683,18 @@ async def serve_modern( if isinstance(route, InboundLadderRejection): raise MCPError(code=route.code, message=route.message, data=route.data) # Classification succeeded: 2026 wire semantics govern this request's - # lifecycle from here. Both facts attach synchronously - no checkpoint - # since the dispatcher entered the handler scope - so a peer cancel - # can never interleave and observe the legacy defaults. - suppress_cancel_answer(dctx) - observe_success_frames(dctx, partial(commit_modern, dctx, route.protocol_version)) + # lifecycle from here. On a still-unlocked connection this is the + # silence rule's exact scope - a request that fails classification + # above has not proven modern semantics and keeps the legacy cancel + # answer - and the era lock rides the request's first success frame. + # Both facts attach synchronously - no checkpoint since the + # dispatcher entered the handler scope - so a peer cancel can never + # interleave and observe the legacy defaults. Once the connection is + # locked, neither attaches here: `on_request` owns the + # connection-scoped silence fact, and the lock can never move again. + if era == "unlocked": + suppress_cancel_answer(dctx) + observe_success_frames(dctx, partial(commit_modern, dctx, route.protocol_version)) connection = Connection.from_envelope( route.protocol_version, route.client_info, @@ -718,6 +736,16 @@ async def on_request( # METHOD_NOT_FOUND a handshake-only server produced, byte for byte. return await loop_runner.on_request(dctx, method, params) if era == "modern": + # Connection-scoped silence: once locked modern, EVERY request is + # governed by the 2026 cancellation rule, rejections included - a + # rejection error is a legitimate answer, but nothing may follow + # the peer's `notifications/cancelled`. Pre-classification rejects + # run on a spawned task, so an adjacent cancel can precede them + # under trio's randomized scheduling (asyncio's FIFO always starts + # the reject's answer write first - the legitimate + # already-answering case; inline `initialize` has no window at + # all). See the cancellation paragraph on `serve_dual_era_loop`. + suppress_cancel_answer(dctx) if method == "initialize": raise MCPError( code=UNSUPPORTED_PROTOCOL_VERSION, diff --git a/tests/server/test_runner.py b/tests/server/test_runner.py index 4b1de10002..4d37ff74e0 100644 --- a/tests/server/test_runner.py +++ b/tests/server/test_runner.py @@ -2101,6 +2101,47 @@ async def list_tools(ctx: Ctx, params: PaginatedRequestParams | None) -> ListToo assert answer.error.code == UNSUPPORTED_PROTOCOL_VERSION +@pytest.mark.anyio +async def test_dual_era_loop_modern_locked_envelope_less_reject_with_adjacent_cancel_never_answers_legacy(): + """On a modern-locked connection the silence commitment is + connection-scoped: it attaches to every request, including one rejected + before classification (envelope-less). With the cancel adjacent behind + the request, asyncio's FIFO scheduling always starts the reject's answer + write before the read loop consumes the cancel - the legitimate + already-answering case - so exactly one frame appears for the id: the + INVALID_PARAMS classification error, never a code-0 legacy cancel answer, + and nothing after it. (The silenced arm - the cancel consumed before the + reject computes, reachable under trio's randomized scheduling - is pinned + at the dispatcher level by + test_suppress_cancel_answer_silences_even_an_error_raised_after_the_cancel_landed.)""" + + async def list_tools(ctx: Ctx, params: PaginatedRequestParams | None) -> ListToolsResult: + return ListToolsResult(tools=[Tool(name="t", input_schema={"type": "object"})]) + + srv: SrvT = Server(name="fast-server", version="0.0.1", on_list_tools=list_tools) + async with raw_dual_era_loop(srv) as (c2s, s2c): + # Lock modern with a classified request. + c2s.send_nowait(_raw_req(1, "tools/list", _modern_params())) + locked = (await s2c.receive()).message + assert isinstance(locked, JSONRPCResponse) + # Envelope-less request with the cancel adjacent behind it. + c2s.send_nowait(_raw_req(9, "tools/list", {"plain": True})) + c2s.send_nowait(_raw_note("notifications/cancelled", {"requestId": 9})) + answer = (await s2c.receive()).message + assert isinstance(answer, JSONRPCError) + assert answer.id == 9 + assert answer.error.code == INVALID_PARAMS + assert answer.error != ErrorData(code=0, message="Request cancelled") + # Nothing further for the cancelled id, and the connection stays healthy. + await anyio.wait_all_tasks_blocked() + with pytest.raises(anyio.WouldBlock): + s2c.receive_nowait() + c2s.send_nowait(_raw_req(10, "tools/list", _modern_params())) + result = (await s2c.receive()).message + assert isinstance(result, JSONRPCResponse) + assert result.id == 10 + + @pytest.mark.anyio async def test_dual_era_loop_maps_unmapped_handler_exceptions_like_the_modern_http_entry(): """An unmapped handler exception on a modern request surfaces as the From 927b48e17b2ec7f85a95fc94a74f25a5d1b29d61 Mon Sep 17 00:00:00 2001 From: Max Isbey <224885523+maxisbey@users.noreply.github.com> Date: Thu, 16 Jul 2026 22:26:27 +0000 Subject: [PATCH 6/7] Tighten the dispatcher seam edges Four small hardenings around the per-request seams: - on_success_frame is invoked through fire_success_frame, which contains a raising observer (log and continue, the same boundary as the subscription bus's listeners) so an observer bug cannot convert the success it is observing into an error response. The only observer today cannot raise; this makes the contract structural. - suppress_cancel_answer / observe_success_frames now reach through NoServerRequestsDispatchContext instead of silently no-oping on it: the wrapper delegates wire I/O to the loop context it wraps, so per-request facts attached through it must land there. The wrapper moves from server/runner.py to shared/dispatcher.py - it was never server-specific, and the shared setters can only name it without a layering violation from there. The no-op remains for genuinely structurally-silent contexts (single-exchange HTTP, direct dispatch). - The legacy code-0 cancel answer template is never written to the wire: the one site that writes a cancel answer sends a fresh model_copy. ErrorData is mutable and frames cross in-memory transports by reference, so sharing the instance would let a consumer's mutation corrupt every later cancel answer process-wide. - ListenHandler's class docstring states the actual transport requirement (any dispatch context that forwards request-scoped notifications: HTTP SSE and the stream-pair dual-era loop) instead of naming SSE alone. --- src/mcp/server/runner.py | 72 +++--------- src/mcp/server/subscriptions.py | 7 +- src/mcp/shared/dispatcher.py | 57 ++++++++++ src/mcp/shared/jsonrpc_dispatcher.py | 93 ++++++++++++---- tests/server/test_runner.py | 7 +- tests/shared/test_jsonrpc_dispatcher.py | 139 +++++++++++++++++++++++- 6 files changed, 284 insertions(+), 91 deletions(-) diff --git a/src/mcp/server/runner.py b/src/mcp/server/runner.py index 037ed23b11..a453ebe717 100644 --- a/src/mcp/server/runner.py +++ b/src/mcp/server/runner.py @@ -15,7 +15,7 @@ import logging from collections.abc import Awaitable, Mapping -from dataclasses import KW_ONLY, dataclass, replace +from dataclasses import KW_ONLY, dataclass from functools import cached_property, partial from typing import TYPE_CHECKING, Any, Generic, Literal, cast @@ -35,7 +35,6 @@ Implementation, InitializeRequestParams, InitializeResult, - RequestId, RequestParams, RequestParamsMeta, UnsupportedProtocolVersionErrorData, @@ -56,8 +55,14 @@ from mcp.server.models import InitializationOptions from mcp.server.session import ServerSession 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 +from mcp.shared.dispatcher import ( + DispatchContext, + Dispatcher, + NoServerRequestsDispatchContext, + OnNotify, + OnRequest, +) +from mcp.shared.exceptions import MCPError from mcp.shared.inbound import InboundLadderRejection, classify_inbound_request from mcp.shared.jsonrpc_dispatcher import ( JSONRPCDispatcher, @@ -65,7 +70,7 @@ observe_success_frames, suppress_cancel_answer, ) -from mcp.shared.message import MessageMetadata, ServerMessageMetadata, SessionMessage +from mcp.shared.message import ServerMessageMetadata, SessionMessage from mcp.shared.transport_context import TransportContext if TYPE_CHECKING: @@ -490,57 +495,6 @@ def modern_error_data(exc: Exception) -> ErrorData: return ErrorData(code=INTERNAL_ERROR, message="Internal server error") -@dataclass -class _NoServerRequestsDispatchContext: - """Delegating `DispatchContext` that refuses server-initiated requests. - - Wraps the loop dispatcher's per-message context for modern-era dispatch: - the modern protocol forbids server-initiated JSON-RPC requests, so - `send_raw_request` refuses while notifications and progress still ride - the duplex pipe. - """ - - _inner: DispatchContext[TransportContext] - - @property - def transport(self) -> TransportContext: - # Mask the per-message flag so the transport metadata agrees with this - # wrapper's denial: the modern HTTP entry builds its context with - # can_send_request=False, while the loop's default builder says True. - transport = self._inner.transport - return replace(transport, can_send_request=False) if transport.can_send_request else transport - - @property - def can_send_request(self) -> bool: - return False - - @property - def request_id(self) -> RequestId | None: - return self._inner.request_id - - @property - def message_metadata(self) -> MessageMetadata: - return self._inner.message_metadata - - @property - def cancel_requested(self) -> anyio.Event: - return self._inner.cancel_requested - - async def send_raw_request( - self, - method: str, - params: Mapping[str, Any] | None, - opts: CallOptions | None = None, - ) -> dict[str, Any]: - raise NoBackChannelError(method) - - async def notify(self, method: str, params: Mapping[str, Any] | None, opts: CallOptions | None = None) -> None: - await self._inner.notify(method, params, opts) - - async def progress(self, progress: float, total: float | None = None, message: str | None = None) -> None: - await self._inner.progress(progress, total, message) - - async def serve_dual_era_loop( server: Server[LifespanT], read_stream: ReadStream[SessionMessage | Exception], @@ -704,7 +658,7 @@ async def serve_modern( try: return await serve_one( server, - _NoServerRequestsDispatchContext(dctx), + NoServerRequestsDispatchContext(dctx), method, params, connection=connection, @@ -772,7 +726,7 @@ async def on_notify(dctx: DispatchContext[TransportContext], method: str, params connection = Connection.from_envelope(modern_version, None, None, outbound=standalone_outbound) notify_runner = ServerRunner(server, connection, lifespan_state) try: - await notify_runner.on_notify(_NoServerRequestsDispatchContext(dctx), method, params) + await notify_runner.on_notify(NoServerRequestsDispatchContext(dctx), method, params) finally: await aclose_shielded(connection) @@ -832,7 +786,7 @@ async def handle( ) return await serve_one( server, - _NoServerRequestsDispatchContext(dctx), + NoServerRequestsDispatchContext(dctx), method, params, connection=connection, diff --git a/src/mcp/server/subscriptions.py b/src/mcp/server/subscriptions.py index 718d26eaf9..4e07ecde10 100644 --- a/src/mcp/server/subscriptions.py +++ b/src/mcp/server/subscriptions.py @@ -164,9 +164,10 @@ 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, or a duplex stream pair (stdio) where the listen request - simply stays pending while notifications ride the pipe. + Serves any transport whose dispatch context forwards request-scoped + notifications: streamable HTTP's SSE mode (the response IS the stream) + and the stream-pair dual-era loop (stdio), where the listen request + simply stays pending while the ack and events ride the duplex pipe. `max_subscriptions` bounds concurrent streams (further listen requests are rejected with `INTERNAL_ERROR`, before the ack). `max_buffered_events` diff --git a/src/mcp/shared/dispatcher.py b/src/mcp/shared/dispatcher.py index f109638f2a..bf34afd402 100644 --- a/src/mcp/shared/dispatcher.py +++ b/src/mcp/shared/dispatcher.py @@ -18,12 +18,14 @@ import logging from collections.abc import Awaitable, Callable, Mapping +from dataclasses import dataclass, replace from typing import Any, Protocol, TypedDict, TypeVar, runtime_checkable import anyio import anyio.abc from mcp_types import RequestId +from mcp.shared.exceptions import NoBackChannelError from mcp.shared.message import MessageMetadata from mcp.shared.transport_context import TransportContext @@ -33,6 +35,7 @@ "CallOptions", "DispatchContext", "Dispatcher", + "NoServerRequestsDispatchContext", "OnNotify", "OnNotifyIntercept", "OnRequest", @@ -218,6 +221,60 @@ async def progress(self, progress: float, total: float | None = None, message: s ... +@dataclass +class NoServerRequestsDispatchContext: + """Delegating `DispatchContext` that refuses server-initiated requests. + + Wraps another dispatch context for entries whose protocol forbids + server-initiated JSON-RPC requests (the modern 2026-07-28 era): + `send_raw_request` refuses while notifications and progress still ride + the wrapped context's channel. Because it delegates wire I/O, the + per-request seam setters in `mcp.shared.jsonrpc_dispatcher` + (`suppress_cancel_answer`, `observe_success_frames`) reach through it to + the wrapped context. + """ + + inner: DispatchContext[TransportContext] + + @property + def transport(self) -> TransportContext: + # Mask the per-message flag so the transport metadata agrees with this + # wrapper's denial: the modern HTTP entry builds its context with + # can_send_request=False, while the loop's default builder says True. + transport = self.inner.transport + return replace(transport, can_send_request=False) if transport.can_send_request else transport + + @property + def can_send_request(self) -> bool: + return False + + @property + def request_id(self) -> RequestId | None: + return self.inner.request_id + + @property + def message_metadata(self) -> MessageMetadata: + return self.inner.message_metadata + + @property + def cancel_requested(self) -> anyio.Event: + return self.inner.cancel_requested + + async def send_raw_request( + self, + method: str, + params: Mapping[str, Any] | None, + opts: CallOptions | None = None, + ) -> dict[str, Any]: + raise NoBackChannelError(method) + + async def notify(self, method: str, params: Mapping[str, Any] | None, opts: CallOptions | None = None) -> None: + await self.inner.notify(method, params, opts) + + async def progress(self, progress: float, total: float | None = None, message: str | None = None) -> None: + await self.inner.progress(progress, total, message) + + OnRequest = Callable[[DispatchContext[TransportContext], str, Mapping[str, Any] | None], Awaitable[dict[str, Any]]] """Handler for inbound requests: `(ctx, method, params) -> result`. Raise `MCPError` to send an error response.""" diff --git a/src/mcp/shared/jsonrpc_dispatcher.py b/src/mcp/shared/jsonrpc_dispatcher.py index 76d78c2e06..0f690e7608 100644 --- a/src/mcp/shared/jsonrpc_dispatcher.py +++ b/src/mcp/shared/jsonrpc_dispatcher.py @@ -43,6 +43,7 @@ CallOptions, DispatchContext, Dispatcher, + NoServerRequestsDispatchContext, OnNotify, OnNotifyIntercept, OnRequest, @@ -134,11 +135,15 @@ class _InFlight(Generic[TransportT]): _LEGACY_CANCEL_ANSWER: Final[ErrorData] = ErrorData(code=0, message="Request cancelled") -"""The answer a peer-cancelled request gets by default. +"""Template for the answer a peer-cancelled request gets by default. TODO(L38): the spec says receivers SHOULD NOT respond after a cancel; the -existing server always has, so this pins released-client compat. Never -mutated - one shared instance is safe. +existing server always has, so this pins released-client compat. A template, +never written to the wire itself: the one site that writes a cancel answer +sends a fresh `model_copy` (`ErrorData` is not frozen, and over in-memory +transports the answer object crosses to the peer by reference, so sharing +one instance would let a consumer's mutation corrupt every later cancel +answer process-wide). """ @@ -158,8 +163,9 @@ class _JSONRPCDispatchContext(Generic[TransportT]): """The error written when a peer cancel interrupts this request's handler. `None` means silence: no frame at all follows the cancel. The default is - the legacy compat answer; loop layers that know a request is governed by - 2026-era wire rules (which forbid any frame for a request after + the shared legacy compat template (the write site sends a copy, never the + template itself); loop layers that know a request is governed by 2026-era + wire rules (which forbid any frame for a request after `notifications/cancelled`) clear it via `suppress_cancel_answer`. """ on_success_frame: Callable[[], None] | None = None @@ -168,7 +174,8 @@ class _JSONRPCDispatchContext(Generic[TransportT]): those) or the success result. Never invoked for error responses, nor for notifications dropped because the context closed. Set via `observe_success_frames`; loop layers use it to commit connection state no - later than the request's first client-visible output. + later than the request's first client-visible output. Always invoked + through `fire_success_frame`, which contains a raising observer. """ @property @@ -193,12 +200,25 @@ def must_stay_silent(self) -> bool: """ return self.cancel_answer is None and self.cancel_requested.is_set() + def fire_success_frame(self) -> None: + """Invoke the success-frame observer ahead of a non-error frame write. + + Contained like the subscription bus's listener boundary: a raising + observer is logged and skipped, never propagated, so an observer bug + cannot convert the success it is observing into an error response. + """ + if self.on_success_frame is None: + return + try: + self.on_success_frame() + except Exception: # observer boundary: the frame must still be written + logger.exception("on_success_frame observer raised; continuing") + async def notify(self, method: str, params: Mapping[str, Any] | None, opts: CallOptions | None = None) -> None: if self._closed: logger.debug("dropped %s: dispatch context closed", method) return - if self.on_success_frame is not None: - self.on_success_frame() + self.fire_success_frame() await self._dispatcher.notify(method, params, opts, _related_request_id=self._request_id) async def send_raw_request( @@ -225,6 +245,23 @@ def close(self) -> None: self._closed = True +def _wire_context(dctx: DispatchContext[Any]) -> _JSONRPCDispatchContext[Any] | None: + """Resolve `dctx` to the JSON-RPC loop context that owns its wire request, if any. + + `NoServerRequestsDispatchContext` delegates wire I/O to the context it + wraps, so per-request facts attached through it must land on the wrapped + context - the seam setters peel it here rather than silently no-oping on + a context that IS a loop request underneath. Every other context type + resolves to None: those are structurally silent after a peer cancel + (single-exchange HTTP unwinds the response stream, direct dispatch + unwinds into the caller's own cancelled scope), so there is no late + answer to configure and the setters are documented no-ops for them. + """ + while isinstance(dctx, NoServerRequestsDispatchContext): + dctx = dctx.inner + return dctx if isinstance(dctx, _JSONRPCDispatchContext) else None + + def suppress_cancel_answer(dctx: DispatchContext[Any]) -> None: """Commit `dctx`'s request to silence on peer cancellation. @@ -234,13 +271,15 @@ def suppress_cancel_answer(dctx: DispatchContext[Any]) -> None: those semantics; do so before the request's first checkpoint, so a racing cancel can never observe the legacy default answer. - No-op for dispatch contexts that are already structurally silent after a - cancel (single-exchange HTTP closes the response stream; direct dispatch - unwinds into the caller's own cancelled scope) - only the JSON-RPC loop - context writes a late answer to suppress. + Reaches through `NoServerRequestsDispatchContext` to the loop context it + delegates to. No-op for dispatch contexts that are already structurally + silent after a cancel (single-exchange HTTP closes the response stream; + direct dispatch unwinds into the caller's own cancelled scope) - only the + JSON-RPC loop context writes a late answer to suppress. """ - if isinstance(dctx, _JSONRPCDispatchContext): - dctx.cancel_answer = None + wire = _wire_context(dctx) + if wire is not None: + wire.cancel_answer = None def observe_success_frames(dctx: DispatchContext[Any], observer: Callable[[], None]) -> None: @@ -250,14 +289,19 @@ def observe_success_frames(dctx: DispatchContext[Any], observer: Callable[[], No success result, in the writing task with no checkpoint before the write - so state the observer commits is settled before the peer can read the frame. It does not fire for error responses or for notifications dropped - because the request already finished. `observer` must not raise. - - No-op for dispatch contexts other than the JSON-RPC loop's: the only - caller is the dual-era loop's era lock, which has no analogue on the + because the request already finished. A raising observer is contained - + logged and skipped, the frame still written - matching the subscription + bus's listener boundary, so an observer bug cannot convert a success into + an error response. + + Reaches through `NoServerRequestsDispatchContext` like + `suppress_cancel_answer`. No-op for other non-loop dispatch contexts: the + only caller is the dual-era loop's era lock, which has no analogue on the single-exchange or direct paths. """ - if isinstance(dctx, _JSONRPCDispatchContext): - dctx.on_success_frame = observer + wire = _wire_context(dctx) + if wire is not None: + wire.on_success_frame = observer def _default_transport_builder(_meta: MessageMetadata) -> TransportContext: @@ -795,8 +839,7 @@ async def _handle_request( # peers drop late responses, while a second answer for one id # would break JSON-RPC. answer_write_started = True - if dctx.on_success_frame is not None: - dctx.on_success_frame() + dctx.fire_success_frame() await self._write_result(req.id, result) if scope.cancelled_caught and dctx.cancel_answer is not None: # anyio absorbs the scope's own cancel at __exit__, and @@ -804,9 +847,11 @@ async def _handle_request( # result write above did not happen - no double response. # `cancel_answer` is the legacy compat answer unless the loop # layer committed this request to 2026 cancel semantics - # (silence) via `suppress_cancel_answer`. + # (silence) via `suppress_cancel_answer`. Written as a fresh + # copy: frames cross in-memory transports by reference, and + # the default answer is a shared template. answer_write_started = True - await self._write_error(req.id, dctx.cancel_answer) + await self._write_error(req.id, dctx.cancel_answer.model_copy()) except anyio.get_cancelled_exc_class(): # Shutdown: answer the request so the peer isn't left waiting - unless # an answer write already started (it may have reached the transport; diff --git a/tests/server/test_runner.py b/tests/server/test_runner.py index 4d37ff74e0..d3957c26fc 100644 --- a/tests/server/test_runner.py +++ b/tests/server/test_runner.py @@ -63,7 +63,6 @@ _extract_meta, _has_modern_envelope, _initialize_after_modern_data, - _NoServerRequestsDispatchContext, aclose_shielded, serve_connection, serve_dual_era_loop, @@ -71,7 +70,7 @@ ) from mcp.server.session import ServerSession from mcp.server.subscriptions import InMemorySubscriptionBus, ListenHandler -from mcp.shared.dispatcher import CallOptions, DispatchContext, OnNotify +from mcp.shared.dispatcher import CallOptions, DispatchContext, NoServerRequestsDispatchContext, OnNotify from mcp.shared.exceptions import MCPError, NoBackChannelError from mcp.shared.jsonrpc_dispatcher import JSONRPCDispatcher from mcp.shared.message import MessageMetadata, SessionMessage @@ -2257,7 +2256,7 @@ async def progress(self, progress: float, total: float | None = None, message: s @pytest.mark.anyio async def test_no_server_requests_dispatch_context_denies_requests_and_delegates_the_rest(): inner = _RecordingInnerDctx() - wrapper = _NoServerRequestsDispatchContext(inner) + wrapper = NoServerRequestsDispatchContext(inner) assert wrapper.can_send_request is False # The transport metadata is masked to agree with the wrapper's denial. assert wrapper.transport == TransportContext(kind="jsonrpc", can_send_request=False) @@ -2275,7 +2274,7 @@ async def test_no_server_requests_dispatch_context_denies_requests_and_delegates def test_no_server_requests_dispatch_context_passes_an_already_denying_transport_through(): inner = _RecordingInnerDctx() inner.transport = TransportContext(kind="jsonrpc", can_send_request=False) - assert _NoServerRequestsDispatchContext(inner).transport is inner.transport + assert NoServerRequestsDispatchContext(inner).transport is inner.transport @pytest.mark.anyio diff --git a/tests/shared/test_jsonrpc_dispatcher.py b/tests/shared/test_jsonrpc_dispatcher.py index bec38992f2..74e8d0ab0d 100644 --- a/tests/shared/test_jsonrpc_dispatcher.py +++ b/tests/shared/test_jsonrpc_dispatcher.py @@ -34,7 +34,13 @@ from mcp.server import Server, ServerRequestContext from mcp.shared._compat import resync_tracer from mcp.shared._context_streams import ContextReceiveStream, ContextSendStream -from mcp.shared.dispatcher import CallOptions, DispatchContext, coerce_request_id +from mcp.shared.dispatcher import ( + CallOptions, + DispatchContext, + Dispatcher, + NoServerRequestsDispatchContext, + coerce_request_id, +) from mcp.shared.exceptions import MCPError, NoBackChannelError from mcp.shared.jsonrpc_dispatcher import ( # pyright: ignore[reportPrivateUsage] JSONRPCDispatcher, @@ -308,6 +314,85 @@ async def on_request(ctx: DCtx, method: str, params: Mapping[str, Any] | None) - assert await client.send_raw_request("m", None) == {"ok": True} +@pytest.mark.anyio +async def test_seam_setters_reach_through_the_no_server_requests_wrapper(): + """`NoServerRequestsDispatchContext` is not structurally silent - it + delegates wire I/O to the loop context it wraps - so facts attached + through it must land on the wrapped context instead of silently no-oping: + the observer fires for the wrapped request's frames, and the silence + commitment suppresses its late cancel answer.""" + calls: list[str] = [] + handler_started = anyio.Event() + + async def on_request(ctx: DCtx, method: str, params: Mapping[str, Any] | None) -> dict[str, Any]: + wrapper = NoServerRequestsDispatchContext(ctx) + if method == "probe": + return {"alive": True} + if method == "emit": + observe_success_frames(wrapper, lambda: calls.append(method)) + return {"ok": True} + suppress_cancel_answer(wrapper) + handler_started.set() + await anyio.sleep_forever() + raise NotImplementedError + + outcome: list[Any] = [] + + async def call_slow(client: Dispatcher[TransportContext], abandoned: anyio.CancelScope) -> None: + # A (buggy) late "Request cancelled" frame would raise MCPError here + # and fail the test through the task group; a late result lands in + # `outcome`. Silence leaves the call parked until `abandoned` is + # cancelled. + with abandoned: + outcome.append(await client.send_raw_request("slow", None, {"request_id": 9})) + + async with running_pair(jsonrpc_pair, server_on_request=on_request) as (client, *_): + with anyio.fail_after(5): + assert await client.send_raw_request("emit", None) == {"ok": True} + # The observer landed on the wrapped loop context: it fired for + # the result frame despite being attached via the wrapper. + assert calls == ["emit"] + abandoned = anyio.CancelScope() + async with anyio.create_task_group() as tg: # pragma: no branch + tg.start_soon(call_slow, client, abandoned) + await handler_started.wait() + await client.notify("notifications/cancelled", {"requestId": 9}) + # Ordering marker: the probe's answer proves the server + # consumed the cancel and wrote nothing for id 9 before it. + assert await client.send_raw_request("probe", None) == {"alive": True} + await anyio.wait_all_tasks_blocked() + abandoned.cancel() + assert outcome == [] + + +@pytest.mark.anyio +async def test_observe_success_frames_contains_a_raising_observer_and_still_writes_the_frames( + caplog: pytest.LogCaptureFixture, +): + """The observer boundary matches the subscription bus's listener boundary: + a raising observer is logged and skipped, and both the request-scoped + notification and the success result still reach the wire - an observer + bug cannot convert the success it is observing into an error response.""" + + def bad_observer() -> None: + raise RuntimeError("observer bug") + + async def on_request(ctx: DCtx, method: str, params: Mapping[str, Any] | None) -> dict[str, Any]: + observe_success_frames(ctx, bad_observer) + await ctx.notify("x/note", None) + return {"ok": True} + + async with running_pair(jsonrpc_pair, server_on_request=on_request) as (client, _server, client_rec, _srec): + with anyio.fail_after(5): + with caplog.at_level(logging.ERROR, logger="mcp.shared.jsonrpc_dispatcher"): + assert await client.send_raw_request("emit", None) == {"ok": True} + await client_rec.notified.wait() + # The observer raised before BOTH frames (notification, then result) and + # both were still written. + assert client_rec.notifications == [("x/note", None)] + assert [r.message for r in caplog.records].count("on_success_frame observer raised; continuing") == 2 + + @pytest.mark.anyio async def test_shutdown_drain_stays_silent_for_a_peer_cancelled_silence_committed_request(): """Loop shutdown catching a silence-committed request whose peer already @@ -366,6 +451,58 @@ async def on_notify(ctx: DCtx, method: str, params: Mapping[str, Any] | None) -> assert late == [] +@pytest.mark.anyio +async def test_legacy_cancel_answer_is_minted_per_request_so_mutation_cannot_corrupt_later_answers(): + """Over in-memory transports the answer object crosses to the peer by + reference, so the legacy cancel answer must be minted per request: a + consumer mutating one received answer must not corrupt the byte-identical + code-0 answer later cancelled requests depend on.""" + c2s_send, c2s_recv = anyio.create_memory_object_stream[SessionMessage | Exception](32) + s2c_send, s2c_recv = anyio.create_memory_object_stream[SessionMessage | Exception](32) + server: JSONRPCDispatcher[TransportContext] = JSONRPCDispatcher(c2s_recv, s2c_send) + started: dict[RequestId, anyio.Event] = {1: anyio.Event(), 2: anyio.Event()} + + async def on_request(ctx: DCtx, method: str, params: Mapping[str, Any] | None) -> dict[str, Any]: + assert ctx.request_id is not None + started[ctx.request_id].set() + await anyio.sleep_forever() + raise NotImplementedError + + async def on_notify(ctx: DCtx, method: str, params: Mapping[str, Any] | None) -> None: + pass + + async def cancel_and_receive_answer(rid: int) -> ErrorData: + await c2s_send.send(SessionMessage(message=JSONRPCRequest(jsonrpc="2.0", id=rid, method="slow"))) + await started[rid].wait() + await c2s_send.send( + SessionMessage( + message=JSONRPCNotification(jsonrpc="2.0", method="notifications/cancelled", params={"requestId": rid}) + ) + ) + answer = await s2c_recv.receive() + assert isinstance(answer, SessionMessage) + assert isinstance(answer.message, JSONRPCError) + assert answer.message.id == rid + return answer.message.error + + try: + async with anyio.create_task_group() as tg: + await tg.start(server.run, on_request, on_notify) + with anyio.fail_after(5): + first = await cancel_and_receive_answer(1) + assert first == ErrorData(code=0, message="Request cancelled") + # A consumer scribbling on its received answer object... + first.message = "corrupted" + # ...must not reach the next request's answer. + second = await cancel_and_receive_answer(2) + assert second == ErrorData(code=0, message="Request cancelled") + assert second is not first + tg.cancel_scope.cancel() + finally: + for s in (c2s_send, c2s_recv, s2c_send, s2c_recv): + s.close() + + @pytest.mark.anyio async def test_peer_cancel_landing_after_handlers_last_checkpoint_writes_only_the_result(): """A peer cancel that fails to interrupt the handler writes only the result: one answer per From 77c35e7d331d3eef7a056ae9821eedd05adcb29d Mon Sep 17 00:00:00 2001 From: Max Isbey <224885523+maxisbey@users.noreply.github.com> Date: Thu, 16 Jul 2026 22:27:12 +0000 Subject: [PATCH 7/7] Convert the remaining listen-over-loop review probes into regression tests Falsy request id 0 through the full listen lifecycle (ack, era lock, and graceful-close result all stamped with the literal 0), double-cancel and stale-cancel idempotence with the max_subscriptions slot recycled, event fan-out surviving the cancellation of a sibling stream, and a plain async-function listen handler whose schema-invalid result fails loudly without locking the era while a valid one serves and locks. --- tests/server/test_runner.py | 135 ++++++++++++++++++++++++++++++++++++ 1 file changed, 135 insertions(+) diff --git a/tests/server/test_runner.py b/tests/server/test_runner.py index d3957c26fc..833bb4c1b0 100644 --- a/tests/server/test_runner.py +++ b/tests/server/test_runner.py @@ -42,6 +42,8 @@ RequestId, RequestParams, SetLevelRequestParams, + SubscriptionsListenRequestParams, + SubscriptionsListenResult, Tool, UnsupportedProtocolVersionErrorData, ) @@ -2141,6 +2143,139 @@ async def list_tools(ctx: Ctx, params: PaginatedRequestParams | None) -> ListToo assert result.id == 10 +@pytest.mark.anyio +async def test_dual_era_loop_listen_with_request_id_zero_stamps_and_resolves_the_falsy_id(): + """Request id 0 is falsy but valid: the ack and the graceful-close result + are stamped with the literal 0, the pending request keeps its id through + the whole lifecycle, and the ack still locks the era.""" + bus = InMemorySubscriptionBus() + handler = ListenHandler(bus) + srv: SrvT = Server(name="listen-server", version="0.0.1", on_subscriptions_listen=handler) + on_notify, notes_send, notes = _note_collector() + result_box: list[dict[str, Any]] = [] + try: + async with dual_era_client(srv, client_on_notify=on_notify) as (client, _): + async with anyio.create_task_group() as tg: + + async def open_listen() -> None: + result_box.append( + await client.send_raw_request( + "subscriptions/listen", _modern_listen_params(), {"request_id": 0} + ) + ) + + tg.start_soon(open_listen) + method, params = await notes.receive() + assert method == ACK_METHOD + assert params["_meta"][SUBSCRIPTION_ID_META_KEY] == 0 + with pytest.raises(MCPError) as init_exc: + await client.send_raw_request("initialize", _initialize_params()) + assert init_exc.value.error.code == UNSUPPORTED_PROTOCOL_VERSION + handler.close() + finally: + notes_send.close() + notes.close() + assert result_box[0]["_meta"][SUBSCRIPTION_ID_META_KEY] == 0 + + +@pytest.mark.anyio +async def test_dual_era_loop_double_cancel_and_stale_cancel_are_silent_and_recycle_the_slot(): + """Cancel idempotence over the loop: a double cancel of a live listen is + total silence and frees its max_subscriptions=1 slot; a cancel for an + already-resolved listen id is ignored and the connection keeps serving.""" + bus = InMemorySubscriptionBus() + handler = ListenHandler(bus, max_subscriptions=1) + srv: SrvT = Server(name="listen-server", version="0.0.1", on_subscriptions_listen=handler) + async with raw_dual_era_loop(srv) as (c2s, s2c): + c2s.send_nowait(_raw_req("d1", "subscriptions/listen", _modern_listen_params())) + await _expect_ack(s2c, "d1") + c2s.send_nowait(_raw_note("notifications/cancelled", {"requestId": "d1"})) + c2s.send_nowait(_raw_note("notifications/cancelled", {"requestId": "d1"})) + await anyio.wait_all_tasks_blocked() + with pytest.raises(anyio.WouldBlock): + s2c.receive_nowait() + # The slot is free again: with max_subscriptions=1 a second listen + # could only ack if the cancelled stream fully cleaned up. + c2s.send_nowait(_raw_req("d2", "subscriptions/listen", _modern_listen_params())) + await _expect_ack(s2c, "d2") + handler.close() + result = (await s2c.receive()).message + assert isinstance(result, JSONRPCResponse) + assert result.id == "d2" + # A stale cancel for the resolved id is ignored: no frame for d2, and + # a third listen still serves. + c2s.send_nowait(_raw_note("notifications/cancelled", {"requestId": "d2"})) + c2s.send_nowait(_raw_req("d3", "subscriptions/listen", _modern_listen_params())) + await _expect_ack(s2c, "d3") + await anyio.wait_all_tasks_blocked() + with pytest.raises(anyio.WouldBlock): + s2c.receive_nowait() + + +@pytest.mark.anyio +async def test_dual_era_loop_cancelling_one_of_two_listens_keeps_the_other_flowing(): + """Two live listen streams on one connection: cancelling one is silent for + its id and events keep fanning out to the survivor, stamped with the + survivor's subscription id only.""" + bus = InMemorySubscriptionBus() + handler = ListenHandler(bus) + srv: SrvT = Server(name="listen-server", version="0.0.1", on_subscriptions_listen=handler) + async with raw_dual_era_loop(srv) as (c2s, s2c): + c2s.send_nowait(_raw_req("a", "subscriptions/listen", _modern_listen_params())) + c2s.send_nowait(_raw_req("b", "subscriptions/listen", _modern_listen_params())) + for expected in ("a", "b"): + await _expect_ack(s2c, expected) + c2s.send_nowait(_raw_note("notifications/cancelled", {"requestId": "a"})) + await anyio.wait_all_tasks_blocked() + await bus.publish(ToolsListChanged()) + event = (await s2c.receive()).message + assert isinstance(event, JSONRPCNotification) + assert event.method == "notifications/tools/list_changed" + assert event.params is not None + assert event.params["_meta"][SUBSCRIPTION_ID_META_KEY] == "b" + # One event, one survivor: nothing was written for the cancelled "a". + await anyio.wait_all_tasks_blocked() + with pytest.raises(anyio.WouldBlock): + s2c.receive_nowait() + handler.close() + result = (await s2c.receive()).message + assert isinstance(result, JSONRPCResponse) + assert result.id == "b" + + +@pytest.mark.anyio +async def test_dual_era_loop_custom_listen_handler_serves_and_only_a_valid_result_locks(): + """A plain async function as `on_subscriptions_listen` (not + `ListenHandler`) serves over the loop like any request. A result missing + the required stamped `_meta` fails outbound validation with + INTERNAL_ERROR and - like every failed request - never locks the era; a + schema-valid result serves and locks.""" + + async def bad(ctx: Ctx, params: SubscriptionsListenRequestParams) -> SubscriptionsListenResult: + return SubscriptionsListenResult() # no _meta: schema-invalid + + srv_bad: SrvT = Server(name="listen-server", version="0.0.1", on_subscriptions_listen=bad) + async with dual_era_client(srv_bad) as (client, _): + with pytest.raises(MCPError) as exc_info: + await client.send_raw_request("subscriptions/listen", _modern_listen_params()) + assert exc_info.value.error.code == INTERNAL_ERROR + assert exc_info.value.error.message == "Handler returned an invalid result" + # The failed listen never locked: the legacy handshake still works. + init = await client.send_raw_request("initialize", _initialize_params()) + assert init["protocolVersion"] == LATEST_HANDSHAKE_VERSION + + async def good(ctx: Ctx, params: SubscriptionsListenRequestParams) -> SubscriptionsListenResult: + return SubscriptionsListenResult(_meta={SUBSCRIPTION_ID_META_KEY: ctx.request_id}) + + srv_good: SrvT = Server(name="listen-server", version="0.0.1", on_subscriptions_listen=good) + async with dual_era_client(srv_good) as (client, _): + result = await client.send_raw_request("subscriptions/listen", _modern_listen_params()) + assert result["resultType"] == "complete" + with pytest.raises(MCPError) as init_exc: + await client.send_raw_request("initialize", _initialize_params()) + assert init_exc.value.error.code == UNSUPPORTED_PROTOCOL_VERSION + + @pytest.mark.anyio async def test_dual_era_loop_maps_unmapped_handler_exceptions_like_the_modern_http_entry(): """An unmapped handler exception on a modern request surfaces as the