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.

Author: maxisbey

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]],

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