Tighten the dispatcher seam edges

Four small hardenings around the per-request seams:

- on_success_frame is invoked through fire_success_frame, which contains a
  raising observer (log and continue, the same boundary as the subscription
  bus's listeners) so an observer bug cannot convert the success it is
  observing into an error response. The only observer today cannot raise;
  this makes the contract structural.

- suppress_cancel_answer / observe_success_frames now reach through
  NoServerRequestsDispatchContext instead of silently no-oping on it: the
  wrapper delegates wire I/O to the loop context it wraps, so per-request
  facts attached through it must land there. The wrapper moves from
  server/runner.py to shared/dispatcher.py - it was never server-specific,
  and the shared setters can only name it without a layering violation from
  there. The no-op remains for genuinely structurally-silent contexts
  (single-exchange HTTP, direct dispatch).

- The legacy code-0 cancel answer template is never written to the wire:
  the one site that writes a cancel answer sends a fresh model_copy.
  ErrorData is mutable and frames cross in-memory transports by reference,
  so sharing the instance would let a consumer's mutation corrupt every
  later cancel answer process-wide.

- ListenHandler's class docstring states the actual transport requirement
  (any dispatch context that forwards request-scoped notifications: HTTP
  SSE and the stream-pair dual-era loop) instead of naming SSE alone.

Author: maxisbey

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,16 +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,
     observe_success_frames,
     suppress_cancel_answer,
 )
-from mcp.shared.message import MessageMetadata, ServerMessageMetadata, SessionMessage
+from mcp.shared.message import ServerMessageMetadata, SessionMessage
 from mcp.shared.transport_context import TransportContext
 
 if TYPE_CHECKING:
@@ -490,57 +495,6 @@ def modern_error_data(exc: Exception) -> ErrorData:
     return ErrorData(code=INTERNAL_ERROR, message="Internal server error")
 
 
-@dataclass
-class _NoServerRequestsDispatchContext:
-    """Delegating `DispatchContext` that refuses server-initiated requests.
-
-    Wraps the loop dispatcher's per-message context for modern-era dispatch:
-    the modern protocol forbids server-initiated JSON-RPC requests, so
-    `send_raw_request` refuses while notifications and progress still ride
-    the duplex pipe.
-    """
-
-    _inner: DispatchContext[TransportContext]
-
-    @property
-    def transport(self) -> TransportContext:
-        # Mask the per-message flag so the transport metadata agrees with this
-        # wrapper's denial: the modern HTTP entry builds its context with
-        # can_send_request=False, while the loop's default builder says True.
-        transport = self._inner.transport
-        return replace(transport, can_send_request=False) if transport.can_send_request else transport
-
-    @property
-    def can_send_request(self) -> bool:
-        return False
-
-    @property
-    def request_id(self) -> RequestId | None:
-        return self._inner.request_id
-
-    @property
-    def message_metadata(self) -> MessageMetadata:
-        return self._inner.message_metadata
-
-    @property
-    def cancel_requested(self) -> anyio.Event:
-        return self._inner.cancel_requested
-
-    async def send_raw_request(
-        self,
-        method: str,
-        params: Mapping[str, Any] | None,
-        opts: CallOptions | None = None,
-    ) -> dict[str, Any]:
-        raise NoBackChannelError(method)
-
-    async def notify(self, method: str, params: Mapping[str, Any] | None, opts: CallOptions | None = None) -> None:
-        await self._inner.notify(method, params, opts)
-
-    async def progress(self, progress: float, total: float | None = None, message: str | None = None) -> None:
-        await self._inner.progress(progress, total, message)
-
-
 async def serve_dual_era_loop(
     server: Server[LifespanT],
     read_stream: ReadStream[SessionMessage | Exception],
@@ -704,7 +658,7 @@ async def serve_modern(
         try:
             return await serve_one(
                 server,
-                _NoServerRequestsDispatchContext(dctx),
+                NoServerRequestsDispatchContext(dctx),
                 method,
                 params,
                 connection=connection,
@@ -772,7 +726,7 @@ async def on_notify(dctx: DispatchContext[TransportContext], method: str, params
         connection = Connection.from_envelope(modern_version, None, None, outbound=standalone_outbound)
         notify_runner = ServerRunner(server, connection, lifespan_state)
         try:
-            await notify_runner.on_notify(_NoServerRequestsDispatchContext(dctx), method, params)
+            await notify_runner.on_notify(NoServerRequestsDispatchContext(dctx), method, params)
         finally:
             await aclose_shielded(connection)
 
@@ -832,7 +786,7 @@ async def handle(
         )
         return await serve_one(
             server,
-            _NoServerRequestsDispatchContext(dctx),
+            NoServerRequestsDispatchContext(dctx),
             method,
             params,
             connection=connection,

src/mcp/server/subscriptions.py

@@ -164,9 +164,10 @@ class ListenHandler:
     cancels the handler; the stream just ends, per the spec's abrupt-close
     contract) or `close` ends all streams gracefully.
 
-    Requires a transport that can stream a request's response: streamable
-    HTTP's SSE mode, or a duplex stream pair (stdio) where the listen request
-    simply stays pending while notifications ride the pipe.
+    Serves any transport whose dispatch context forwards request-scoped
+    notifications: streamable HTTP's SSE mode (the response IS the stream)
+    and the stream-pair dual-era loop (stdio), where the listen request
+    simply stays pending while the ack and events ride the duplex pipe.
 
     `max_subscriptions` bounds concurrent streams (further listen requests are
     rejected with `INTERNAL_ERROR`, before the ack). `max_buffered_events`

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."""
 

src/mcp/shared/jsonrpc_dispatcher.py

@@ -43,6 +43,7 @@
     CallOptions,
     DispatchContext,
     Dispatcher,
+    NoServerRequestsDispatchContext,
     OnNotify,
     OnNotifyIntercept,
     OnRequest,
@@ -134,11 +135,15 @@ class _InFlight(Generic[TransportT]):
 
 
 _LEGACY_CANCEL_ANSWER: Final[ErrorData] = ErrorData(code=0, message="Request cancelled")
-"""The answer a peer-cancelled request gets by default.
+"""Template for the answer a peer-cancelled request gets by default.
 
 TODO(L38): the spec says receivers SHOULD NOT respond after a cancel; the
-existing server always has, so this pins released-client compat. Never
-mutated - one shared instance is safe.
+existing server always has, so this pins released-client compat. A template,
+never written to the wire itself: the one site that writes a cancel answer
+sends a fresh `model_copy` (`ErrorData` is not frozen, and over in-memory
+transports the answer object crosses to the peer by reference, so sharing
+one instance would let a consumer's mutation corrupt every later cancel
+answer process-wide).
 """
 
 
@@ -158,8 +163,9 @@ class _JSONRPCDispatchContext(Generic[TransportT]):
     """The error written when a peer cancel interrupts this request's handler.
 
     `None` means silence: no frame at all follows the cancel. The default is
-    the legacy compat answer; loop layers that know a request is governed by
-    2026-era wire rules (which forbid any frame for a request after
+    the shared legacy compat template (the write site sends a copy, never the
+    template itself); loop layers that know a request is governed by 2026-era
+    wire rules (which forbid any frame for a request after
     `notifications/cancelled`) clear it via `suppress_cancel_answer`.
     """
     on_success_frame: Callable[[], None] | None = None
@@ -168,7 +174,8 @@ class _JSONRPCDispatchContext(Generic[TransportT]):
     those) or the success result. Never invoked for error responses, nor for
     notifications dropped because the context closed. Set via
     `observe_success_frames`; loop layers use it to commit connection state no
-    later than the request's first client-visible output.
+    later than the request's first client-visible output. Always invoked
+    through `fire_success_frame`, which contains a raising observer.
     """
 
     @property
@@ -193,12 +200,25 @@ def must_stay_silent(self) -> bool:
         """
         return self.cancel_answer is None and self.cancel_requested.is_set()
 
+    def fire_success_frame(self) -> None:
+        """Invoke the success-frame observer ahead of a non-error frame write.
+
+        Contained like the subscription bus's listener boundary: a raising
+        observer is logged and skipped, never propagated, so an observer bug
+        cannot convert the success it is observing into an error response.
+        """
+        if self.on_success_frame is None:
+            return
+        try:
+            self.on_success_frame()
+        except Exception:  # observer boundary: the frame must still be written
+            logger.exception("on_success_frame observer raised; continuing")
+
     async def notify(self, method: str, params: Mapping[str, Any] | None, opts: CallOptions | None = None) -> None:
         if self._closed:
             logger.debug("dropped %s: dispatch context closed", method)
             return
-        if self.on_success_frame is not None:
-            self.on_success_frame()
+        self.fire_success_frame()
         await self._dispatcher.notify(method, params, opts, _related_request_id=self._request_id)
 
     async def send_raw_request(
@@ -225,6 +245,23 @@ def close(self) -> None:
         self._closed = True
 
 
+def _wire_context(dctx: DispatchContext[Any]) -> _JSONRPCDispatchContext[Any] | None:
+    """Resolve `dctx` to the JSON-RPC loop context that owns its wire request, if any.
+
+    `NoServerRequestsDispatchContext` delegates wire I/O to the context it
+    wraps, so per-request facts attached through it must land on the wrapped
+    context - the seam setters peel it here rather than silently no-oping on
+    a context that IS a loop request underneath. Every other context type
+    resolves to None: those are structurally silent after a peer cancel
+    (single-exchange HTTP unwinds the response stream, direct dispatch
+    unwinds into the caller's own cancelled scope), so there is no late
+    answer to configure and the setters are documented no-ops for them.
+    """
+    while isinstance(dctx, NoServerRequestsDispatchContext):
+        dctx = dctx.inner
+    return dctx if isinstance(dctx, _JSONRPCDispatchContext) else None
+
+
 def suppress_cancel_answer(dctx: DispatchContext[Any]) -> None:
     """Commit `dctx`'s request to silence on peer cancellation.
 
@@ -234,13 +271,15 @@ def suppress_cancel_answer(dctx: DispatchContext[Any]) -> None:
     those semantics; do so before the request's first checkpoint, so a racing
     cancel can never observe the legacy default answer.
 
-    No-op for dispatch contexts that are already structurally silent after a
-    cancel (single-exchange HTTP closes the response stream; direct dispatch
-    unwinds into the caller's own cancelled scope) - only the JSON-RPC loop
-    context writes a late answer to suppress.
+    Reaches through `NoServerRequestsDispatchContext` to the loop context it
+    delegates to. No-op for dispatch contexts that are already structurally
+    silent after a cancel (single-exchange HTTP closes the response stream;
+    direct dispatch unwinds into the caller's own cancelled scope) - only the
+    JSON-RPC loop context writes a late answer to suppress.
     """
-    if isinstance(dctx, _JSONRPCDispatchContext):
-        dctx.cancel_answer = None
+    wire = _wire_context(dctx)
+    if wire is not None:
+        wire.cancel_answer = None
 
 
 def observe_success_frames(dctx: DispatchContext[Any], observer: Callable[[], None]) -> None:
@@ -250,14 +289,19 @@ def observe_success_frames(dctx: DispatchContext[Any], observer: Callable[[], No
     success result, in the writing task with no checkpoint before the write -
     so state the observer commits is settled before the peer can read the
     frame. It does not fire for error responses or for notifications dropped
-    because the request already finished. `observer` must not raise.
-
-    No-op for dispatch contexts other than the JSON-RPC loop's: the only
-    caller is the dual-era loop's era lock, which has no analogue on the
+    because the request already finished. A raising observer is contained -
+    logged and skipped, the frame still written - matching the subscription
+    bus's listener boundary, so an observer bug cannot convert a success into
+    an error response.
+
+    Reaches through `NoServerRequestsDispatchContext` like
+    `suppress_cancel_answer`. No-op for other non-loop dispatch contexts: the
+    only caller is the dual-era loop's era lock, which has no analogue on the
     single-exchange or direct paths.
     """
-    if isinstance(dctx, _JSONRPCDispatchContext):
-        dctx.on_success_frame = observer
+    wire = _wire_context(dctx)
+    if wire is not None:
+        wire.on_success_frame = observer
 
 
 def _default_transport_builder(_meta: MessageMetadata) -> TransportContext:
@@ -795,18 +839,19 @@ async def _handle_request(
                 # peers drop late responses, while a second answer for one id
                 # would break JSON-RPC.
                 answer_write_started = True
-                if dctx.on_success_frame is not None:
-                    dctx.on_success_frame()
+                dctx.fire_success_frame()
                 await self._write_result(req.id, result)
             if scope.cancelled_caught and dctx.cancel_answer is not None:
                 # anyio absorbs the scope's own cancel at __exit__, and
                 # `cancelled_caught` (unlike `cancel_called`) guarantees the
                 # result write above did not happen - no double response.
                 # `cancel_answer` is the legacy compat answer unless the loop
                 # layer committed this request to 2026 cancel semantics
-                # (silence) via `suppress_cancel_answer`.
+                # (silence) via `suppress_cancel_answer`. Written as a fresh
+                # copy: frames cross in-memory transports by reference, and
+                # the default answer is a shared template.
                 answer_write_started = True
-                await self._write_error(req.id, dctx.cancel_answer)
+                await self._write_error(req.id, dctx.cancel_answer.model_copy())
         except anyio.get_cancelled_exc_class():
             # Shutdown: answer the request so the peer isn't left waiting - unless
             # an answer write already started (it may have reached the transport;