Harden the listen driver against receive-order races and hostile peers

Move the client's listen-stream demux to a synchronous notification
intercept on the dispatcher's receive path, where cancelled/progress
interception already lives: acks, events, and teardown signals now
advance in wire order relative to the listen result, so a graceful
close can no longer outrun the events that preceded it or clobber the
acknowledged filter with the fabricated empty one. Ack consumption
keys on the live route registry alone (the per-session id set is
gone), route admission enforces the honored filter - loose on URIs,
which the spec lets be sub-resources, with a capped backlog settling
the route lost against floods - and the response-cache eviction
barrier moved to the consumption point, held pending until it
completes so a cancelled consumer cannot lose a level trigger.

On the transport, a request whose SSE stream can never answer - a
non-resumable drop or an exhausted reconnection budget - now resolves
with a synthesized error, contained against teardown races.

Docs: a graceful close is not "stop watching" (servers shed load by
closing gracefully, including this SDK's ListenHandler); the watch
loop re-listens on both endings, and sub.honored is the delivery
contract, not catalog state.

Author: maxisbey

docs/advanced/subscriptions.md

@@ -88,9 +88,9 @@ Consuming a subscription is one context manager:
 ```
 
 * `client.listen(...)` takes the filter as keyword arguments — they mirror the wire `SubscriptionFilter` field for field. Entering sends the request and returns once the server's acknowledgment arrives, so `sub.honored` (the subset the server agreed to deliver) is always there before the first event.
-* Iteration yields the same four typed events the server publishes: `ToolsListChanged`, `PromptsListChanged`, `ResourcesListChanged`, and `ResourceUpdated(uri=...)`. An event is a cue to refetch — it carries no payload beyond identity, and duplicates pending consumption collapse into one.
+* Iteration yields the same four typed events the server publishes: `ToolsListChanged`, `PromptsListChanged`, `ResourcesListChanged`, and `ResourceUpdated(uri=...)` — where the URI may be a sub-resource of one you subscribed to, at the server's discretion. An event is a cue to refetch — it carries no payload beyond identity, and duplicates pending consumption collapse into one.
 * Leaving the block ends the subscription, with the transport's own spelling: over streamable HTTP the request's response stream is closed (that is the 2026 cancellation signal), on stream transports `notifications/cancelled` is sent.
-* The stream's two endings are control flow. The server closing gracefully simply ends the `async for`; an abrupt drop raises `SubscriptionLost`. There is no replay and no automatic re-listen — a client that reconnects refetches what it depends on:
+* The stream's two endings are control flow. The server closing gracefully simply ends the `async for`; an abrupt drop raises `SubscriptionLost`. The distinction is diagnostic — a clean end versus a connection worth suspecting — not a difference in what to do next: either way the stream is gone, nothing is replayed, and a watcher that still cares re-listens and refetches. Servers close streams gracefully for their own reasons — shutdown, or shedding a subscriber whose backlog grew past bounds, as this SDK's `ListenHandler` does — so a graceful close is not a signal to stop watching:
 
 ```python
 async def watch(client: Client, uri: str) -> None:
@@ -101,12 +101,14 @@ async def watch(client: Client, uri: str) -> None:
                 async for _event in sub:
                     await client.read_resource(uri)
         except SubscriptionLost:
-            continue  # transport dropped - re-listen
-        else:
-            break  # the server ended it deliberately
+            pass
+        # Graceful close or abrupt drop, the stream is gone either way. Back
+        # off before re-listening - a graceful close may be the server
+        # shedding load, and reconnecting instantly recreates the pressure.
+        await anyio.sleep(1)
 ```
 
-* Checking the acknowledgment (the spec's client SHOULD) is reading `sub.honored` — for example, `if not sub.honored.prompts_list_changed:` the server has no prompts to watch. Multiple subscriptions may be open concurrently; each demultiplexes by its own subscription id.
+* Checking the acknowledgment (the spec's client SHOULD) is reading `sub.honored` — the kinds this stream will actually receive. A server may narrow the filter it agrees to honor (a multi-tenant server declining a URI, say), and `sub.honored` is that delivery contract — it says nothing about what exists in the catalog. Multiple subscriptions may be open concurrently; each demultiplexes by its own subscription id.
 * Tool calls and other requests run freely beside an open stream — from the same task between events, or from sibling tasks sharing the client. A watcher task that refetches inside its event loop is the intended pattern, not a re-entrancy hazard.
 * `listen()` requires a 2026-07-28 connection and raises `ListenNotSupportedError` on older ones, steering to the deprecated `subscribe_resource` and `message_handler` spelling those wires use.
 

src/mcp/client/client.py

@@ -58,7 +58,7 @@
     SamplingFnT,
 )
 from mcp.client.streamable_http import streamable_http_client
-from mcp.client.subscriptions import Subscription
+from mcp.client.subscriptions import ServerEvent, Subscription
 from mcp.client.subscriptions import listen as _listen
 from mcp.server import Server
 from mcp.server.mcpserver import MCPServer
@@ -69,6 +69,7 @@
 from mcp.shared.extension import validate_extension_identifier
 from mcp.shared.jsonrpc_dispatcher import JSONRPCDispatcher
 from mcp.shared.session import RequestResponder
+from mcp.shared.subscriptions import event_to_notification
 
 logger = logging.getLogger(__name__)
 
@@ -684,7 +685,8 @@ def listen(
                     tools = await client.list_tools()  # refetch on change
 
         A graceful server close ends the loop; an abrupt drop raises
-        `SubscriptionLost` (re-listen and refetch - there is no replay).
+        `SubscriptionLost`. Either way the stream is gone and there is no
+        replay: a client that keeps watching re-listens and refetches.
         Exiting the context ends the subscription. Multiple subscriptions may
         be open concurrently.
 
@@ -702,8 +704,28 @@ def listen(
             prompts_list_changed=prompts_list_changed,
             resources_list_changed=resources_list_changed,
             resource_subscriptions=resource_subscriptions,
+            on_event=self._evict_for_listen_event if self._response_cache is not None else None,
         )
 
+    async def _evict_for_listen_event(self, event: ServerEvent) -> None:
+        """Finish response-cache eviction before a listen consumer can refetch.
+
+        The eviction wrapper on message_handler runs on a spawned path; the
+        consumer's iterator would otherwise wake first, refetch through a
+        still-warm entry, and - events being deduplicated level triggers -
+        never get a second wake to correct it. The event's notification still
+        tees to that wrapper, so the same eviction fires twice; that is
+        deliberate (eviction is idempotent, and the tee-path one covers
+        non-iterating consumers sharing this cache). Same containment boundary
+        as `_evicting_message_handler`: a cache fault must not block delivery.
+        """
+        cache = self._response_cache
+        assert cache is not None  # installed as the event barrier only when a cache exists
+        try:
+            await cache.evict_for_notification(event_to_notification(event, {}))
+        except Exception:  # boundary: eviction reaches user store code; a cache fault must not block delivery
+            logger.exception("Response cache eviction failed; the event is still delivered")
+
     @deprecated(
         "resources/subscribe is removed as of 2026-07-28; use Client.listen() instead.",
         category=MCPDeprecationWarning,

src/mcp/client/session.py

@@ -50,10 +50,10 @@
     mcp_param_headers,
     x_mcp_header_map,
 )
-from mcp.shared.jsonrpc_dispatcher import JSONRPCDispatcher
+from mcp.shared.jsonrpc_dispatcher import JSONRPCDispatcher, cancelled_request_id_from_params
 from mcp.shared.message import ClientMessageMetadata, SessionMessage
 from mcp.shared.session import RequestResponder
-from mcp.shared.subscriptions import SUBSCRIPTION_ID_META_KEY, event_from_notification
+from mcp.shared.subscriptions import SUBSCRIPTION_ID_META_KEY, event_from_wire
 from mcp.shared.transport_context import TransportContext
 
 DEFAULT_CLIENT_INFO = types.Implementation(name="mcp", version="0.1.0")
@@ -365,11 +365,11 @@ def __init__(
         self._task_group: anyio.abc.TaskGroup | None = None
         # subscriptions/listen demux routes, keyed by the listen request's id
         # (verbatim-typed: a plain dict already keeps 1 and "1" distinct).
+        # Fed by `_intercept_notification` on the dispatcher's receive path;
+        # membership alone decides ack consumption, so a closed subscription
+        # releases its id completely - raw escape-hatch listens (never
+        # registered here) keep receiving their acks via message_handler.
         self._listen_routes: dict[RequestId, ListenRoute] = {}
-        # Every id the driver ever minted on this session: a late ack for a
-        # closed driver listen must be dropped, while raw escape-hatch listens
-        # (never in this set) keep receiving their acks via message_handler.
-        self._driver_listen_ids: set[RequestId] = set()
         if dispatcher is not None:
             if read_stream is not None or write_stream is not None:
                 raise ValueError("pass read_stream/write_stream or dispatcher, not both")
@@ -398,7 +398,9 @@ async def __aenter__(self) -> Self:
             for binding in self._notification_bindings.values():
                 send, receive = anyio.create_memory_object_stream[BaseModel](_NOTIFICATION_QUEUE_SIZE)
                 self._binding_queues[binding.method] = (send, receive)
-            await self._task_group.start(self._dispatcher.run, self._on_request, self._on_notify)
+            await self._task_group.start(
+                self._dispatcher.run, self._on_request, self._on_notify, self._intercept_notification
+            )
             for binding in self._notification_bindings.values():
                 _, receive = self._binding_queues[binding.method]
                 self._task_group.start_soon(self._deliver_bound_notifications, binding, receive)
@@ -1248,7 +1250,6 @@ def _register_listen_route(self, request_id: RequestId) -> ListenRoute:
         """Create the demux route for a listen request id; the caller registers BEFORE sending."""
         route = ListenRoute()
         self._listen_routes[request_id] = route
-        self._driver_listen_ids.add(request_id)
         return route
 
     def _unregister_listen_route(self, request_id: RequestId) -> None:
@@ -1266,21 +1267,60 @@ def _settle_listen_routes_closed(self) -> None:
             route.settle("lost", error=closed)
         self._listen_routes.clear()
 
-    def _listen_subscription_id(self, notification: types.ServerNotification) -> RequestId | None:
-        """The frame's `_meta` subscription id, shape-checked.
-
-        The `as_request_id` guard is not a tripwire: on pre-2026 wires the meta
-        key is untyped, so a non-id (even unhashable) value is constructible
-        and would fail the dict lookup; 2026 surface validation already
-        rejects those shapes.
+    def _intercept_notification(self, method: str, params: Mapping[str, Any] | None) -> bool:
+        """Wire-order listen demux, run by the dispatcher on its receive path.
+
+        Route bookkeeping must advance in receive order relative to the listen
+        request's own result: the result resolves synchronously on this same
+        path, so an ack or event handled on the spawned `_on_notify` path could
+        lose the race and be dropped after the stream settled - a graceful
+        close swallowing the events that preceded it. Only the synchronous
+        bookkeeping happens here; the user-facing tee (message_handler,
+        logging) stays on the spawned path.
+
+        Returns True to consume the frame: an ack for a live driver route is
+        driver state, never surfaced. Raw escape-hatch listens have no route
+        registered and keep observing their acks via message_handler - as does
+        a stray ack for an already-closed driver id, whose registration is gone.
+
+        The `as_request_id` guard is not a tripwire: this reads raw wire
+        dicts, where a non-id (even unhashable) `_meta` value is
+        constructible and would fail the dict lookup.
         """
-        meta = notification.params.meta if notification.params is not None else None
-        return as_request_id(meta.get(SUBSCRIPTION_ID_META_KEY)) if meta is not None else None
-
-    def _listen_route_for(self, notification: types.ServerNotification) -> ListenRoute | None:
-        """The route a listen-stream frame belongs to, by its `_meta` subscription id."""
-        subscription_id = self._listen_subscription_id(notification)
-        return self._listen_routes.get(subscription_id) if subscription_id is not None else None
+        if not self._listen_routes:
+            return False
+        if method == "notifications/cancelled":
+            request_id = cancelled_request_id_from_params(params)
+            if request_id is not None and (listen_route := self._listen_routes.get(request_id)) is not None:
+                # A server-sent cancel naming one of our listen requests is
+                # that stream's teardown signal (the stream-transport spelling).
+                listen_route.settle("lost")
+            return False  # _on_notify swallows every cancelled either way (v1 parity)
+        if params is None:
+            return False
+        meta = params.get("_meta")
+        if not isinstance(meta, Mapping):
+            return False
+        subscription_id = as_request_id(cast("Mapping[str, Any]", meta).get(SUBSCRIPTION_ID_META_KEY))
+        if subscription_id is None or (listen_route := self._listen_routes.get(subscription_id)) is None:
+            return False
+        if method == "notifications/subscriptions/acknowledged":
+            raw_filter = params.get("notifications")
+            if raw_filter is None:
+                # The wire shape requires `notifications`; a frame without it is
+                # malformed, not an empty filter - leave it to the spawned
+                # path's validation warning rather than fabricating honored={}.
+                return False
+            try:
+                honored = types.SubscriptionFilter.model_validate(raw_filter)
+            except ValidationError:
+                # Malformed ack: leave it to the spawned path's validation warning.
+                return False
+            listen_route.set_acked(honored)
+            return True
+        if (event := event_from_wire(method, params)) is not None:
+            listen_route.deliver(event)
+        return False  # events (and any other stamped frame) still tee as usual
 
     async def _on_notify(
         self, dctx: DispatchContext[TransportContext], method: str, params: Mapping[str, Any] | None
@@ -1316,24 +1356,10 @@ async def _on_notify(
             logger.warning("Failed to validate notification: %s", method, exc_info=True)
             return
         if isinstance(notification, types.CancelledNotification):
-            # A server-sent cancel naming one of our listen requests is that
-            # stream's teardown signal (the stream-transport spelling); any
-            # other cancellation was already applied by the dispatcher.
-            # Neither is surfaced to message_handler.
-            cancelled_id = notification.params.request_id
-            if cancelled_id is not None and (listen_route := self._listen_routes.get(cancelled_id)) is not None:
-                listen_route.settle("lost")
+            # Never surfaced to message_handler (v1 parity): the dispatcher
+            # already applied any in-flight cancellation, and a cancel naming
+            # one of our listen streams was settled by the wire-order intercept.
             return
-        if isinstance(notification, types.SubscriptionsAcknowledgedNotification):
-            subscription_id = self._listen_subscription_id(notification)
-            if subscription_id is not None and subscription_id in self._driver_listen_ids:
-                # Driver state, never surfaced: a late ack for an already-closed
-                # driver listen is dropped rather than leaking to message_handler.
-                # Acks outside the driver's id namespace fall through - a raw
-                # escape-hatch listen observes its ack there.
-                if (listen_route := self._listen_routes.get(subscription_id)) is not None:
-                    listen_route.set_acked(notification.params.notifications)
-                return
         try:
             if isinstance(notification, types.LoggingMessageNotification):
                 await self._logging_callback(notification.params)
@@ -1344,14 +1370,6 @@ async def _on_notify(
             # would otherwise fail the peer's send. A raising logging_callback
             # skips the message_handler tee for that notification (v1 parity).
             logger.exception("notification callback for %r raised", method)
-        # Deliver AFTER the tee: the caching layer's eviction wrapper runs in
-        # message_handler, and a consumer woken first could refetch into the
-        # stale entry - with deduplicated level-trigger events there would be
-        # no second wake to correct it.
-        if (listen_route := self._listen_route_for(notification)) is not None:
-            event = event_from_notification(notification)
-            if event is not None:
-                listen_route.deliver(event)
 
     async def _on_stream_exception(self, exc: Exception) -> None:
         """Deliver a transport-level fault to message_handler via a spawned task.

src/mcp/client/streamable_http.py

@@ -447,9 +447,25 @@ async def _handle_sse_response(
             # instead of leaving the request pending for the session's lifetime
             # (a listen stream's consumer would otherwise hang instead of
             # learning the subscription is lost).
-            error_data = ErrorData(code=CONNECTION_CLOSED, message="SSE stream ended without a response")
-            error_msg = SessionMessage(JSONRPCError(jsonrpc="2.0", id=original_request_id, error=error_data))
-            await ctx.read_stream_writer.send(error_msg)
+            await self._resolve_abandoned_request(
+                ctx.read_stream_writer, original_request_id, "SSE stream ended without a response"
+            )
+
+    async def _resolve_abandoned_request(
+        self, read_stream_writer: StreamWriter, request_id: RequestId, message: str
+    ) -> None:
+        """Resolve a request whose response can never arrive with a synthesized error.
+
+        Best-effort, like the dispatcher's own resolution writes: the read
+        stream closing first means the session is being torn down and nobody
+        is waiting on this request anymore.
+        """
+        error_data = ErrorData(code=CONNECTION_CLOSED, message=message)
+        error_msg = SessionMessage(JSONRPCError(jsonrpc="2.0", id=request_id, error=error_data))
+        try:
+            await read_stream_writer.send(error_msg)
+        except (anyio.BrokenResourceError, anyio.ClosedResourceError):
+            logger.debug("read stream closed before request %r could be resolved", request_id)
 
     async def _handle_reconnection(
         self,
@@ -459,9 +475,22 @@ async def _handle_reconnection(
         attempt: int = 0,
     ) -> None:
         """Reconnect with Last-Event-ID to resume stream after server disconnect."""
-        # Bail if max retries exceeded
-        if attempt >= MAX_RECONNECTION_ATTEMPTS:  # pragma: no cover
+        # Only requests reconnect: every caller reaches here from a request's
+        # response stream (`_handle_sse_response` asserts the same), so the
+        # original id to map responses onto is always present.
+        assert isinstance(ctx.session_message.message, JSONRPCRequest)
+        original_request_id = ctx.session_message.message.id
+
+        if attempt >= MAX_RECONNECTION_ATTEMPTS:
+            # Give up AND resolve: without a synthesized error the waiter is
+            # pending for the session's lifetime, and a request with no read
+            # timeout (a listen stream) would hang its caller forever.
             logger.debug(f"Max reconnection attempts ({MAX_RECONNECTION_ATTEMPTS}) exceeded")
+            await self._resolve_abandoned_request(
+                ctx.read_stream_writer,
+                original_request_id,
+                "SSE stream ended and reconnection attempts were exhausted",
+            )
             return
 
         # Always wait - use server value or default
@@ -471,11 +500,6 @@ async def _handle_reconnection(
         headers = self._prepare_headers()
         headers[LAST_EVENT_ID] = last_event_id
 
-        # Extract original request ID to map responses
-        original_request_id = None
-        if isinstance(ctx.session_message.message, JSONRPCRequest):  # pragma: no branch
-            original_request_id = ctx.session_message.message.id
-
         try:
             async with aconnect_sse(ctx.client, "GET", self.url, headers=headers) as event_source:
                 event_source.response.raise_for_status()