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..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,11 +55,22 @@ 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, handler_exception_to_error_data -from mcp.shared.message import MessageMetadata, ServerMessageMetadata, SessionMessage +from mcp.shared.jsonrpc_dispatcher import ( + JSONRPCDispatcher, + handler_exception_to_error_data, + observe_success_frames, + suppress_cancel_answer, +) +from mcp.shared.message import ServerMessageMetadata, SessionMessage from mcp.shared.transport_context import TransportContext if TYPE_CHECKING: @@ -485,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], @@ -575,15 +534,65 @@ 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 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` + the server sends nothing further for that request - no result, no error + (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, @@ -601,28 +610,45 @@ 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: 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 + 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. 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, @@ -630,9 +656,9 @@ async def serve_modern( outbound=standalone_outbound, ) try: - result = await serve_one( + return await serve_one( server, - _NoServerRequestsDispatchContext(dctx), + NoServerRequestsDispatchContext(dctx), method, params, connection=connection, @@ -647,9 +673,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 @@ -667,6 +690,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, @@ -693,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) @@ -753,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 6b0b3d49b5..4e07ecde10 100644 --- a/src/mcp/server/subscriptions.py +++ b/src/mcp/server/subscriptions.py @@ -164,8 +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). + 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 42798fdc54..0f690e7608 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 @@ -43,6 +43,7 @@ CallOptions, DispatchContext, Dispatcher, + NoServerRequestsDispatchContext, OnNotify, OnNotifyIntercept, OnRequest, @@ -64,7 +65,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 +134,19 @@ class _InFlight(Generic[TransportT]): dctx: _JSONRPCDispatchContext[TransportT] +_LEGACY_CANCEL_ANSWER: Final[ErrorData] = ErrorData(code=0, message="Request cancelled") +"""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. 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). +""" + + @dataclass class _JSONRPCDispatchContext(Generic[TransportT]): """Concrete `DispatchContext` produced for each inbound JSON-RPC message.""" @@ -143,6 +159,24 @@ 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 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 + """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. Always invoked + through `fire_success_frame`, which contains a raising observer. + """ @property def request_id(self) -> RequestId | None: @@ -152,10 +186,39 @@ 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() + + 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 + self.fire_success_frame() await self._dispatcher.notify(method, params, opts, _related_request_id=self._request_id) async def send_raw_request( @@ -182,6 +245,65 @@ 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. + + 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. + + 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. + """ + wire = _wire_context(dctx) + if wire is not None: + wire.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. 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. + """ + wire = _wire_context(dctx) + if wire is not None: + wire.on_success_frame = observer + + def _default_transport_builder(_meta: MessageMetadata) -> TransportContext: return TransportContext(kind="jsonrpc", can_send_request=True) @@ -717,21 +839,29 @@ async def _handle_request( # peers drop late responses, while a second answer for one id # would break JSON-RPC. answer_write_started = True + dctx.fire_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`. 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, ErrorData(code=0, message="Request cancelled")) + 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; - # 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, @@ -742,12 +872,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 +901,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.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) + async def _final_write( self, write: Callable[[], Awaitable[None]], diff --git a/tests/server/test_runner.py b/tests/server/test_runner.py index 29d3f07fa6..833bb4c1b0 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, @@ -30,14 +31,21 @@ ErrorData, Implementation, InitializeRequestParams, + JSONRPCError, + JSONRPCNotification, JSONRPCRequest, + JSONRPCResponse, ListToolsResult, NotificationParams, PaginatedRequestParams, ProgressNotificationParams, + RequestId, RequestParams, SetLevelRequestParams, + SubscriptionsListenRequestParams, + SubscriptionsListenResult, Tool, + UnsupportedProtocolVersionErrorData, ) from mcp_types.version import ( LATEST_HANDSHAKE_VERSION, @@ -57,18 +65,19 @@ _extract_meta, _has_modern_envelope, _initialize_after_modern_data, - _NoServerRequestsDispatchContext, aclose_shielded, serve_connection, serve_dual_era_loop, 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, NoServerRequestsDispatchContext, 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 +1263,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 +1284,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 +1429,353 @@ 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 + + +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 + 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 +1926,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,29 +1945,336 @@ 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. + 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 +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_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_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(): @@ -1736,7 +2391,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) @@ -1754,7 +2409,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 @@ -1773,3 +2428,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") diff --git a/tests/shared/test_jsonrpc_dispatcher.py b/tests/shared/test_jsonrpc_dispatcher.py index 5c29c7e3ce..74e8d0ab0d 100644 --- a/tests/shared/test_jsonrpc_dispatcher.py +++ b/tests/shared/test_jsonrpc_dispatcher.py @@ -34,18 +34,26 @@ 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, _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 +164,345 @@ 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_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 + 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_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