Add the client-side subscriptions/listen driver

[maxisbey]

Stacked on #3046 (the request-id and cancellation plumbing) — review this one for the driver only; the base branch carries the transport work.

The client half of subscriptions/listen (2026-07-28, SEP-2575), shaped after the ecosystem's own streaming idioms: one context manager, async for consumption, the acknowledgment as state on the handle rather than an event in the stream.

async with client.listen(tools_list_changed=True, resource_subscriptions=["note://todo"]) as sub:
    print(sub.honored)  # the subset the server agreed to deliver
    async for event in sub:
        match event:
            case ToolsListChanged():
                tools = await client.list_tools()
            case ResourceUpdated(uri=uri):
                body = await client.read_resource(uri)

Motivation and Context

On the 2026 wire the response to subscriptions/listen is the notification stream. The server half shipped in #3035; until now a client had to hand-roll a message_handler, park a raw send_request in a task, and demultiplex _meta subscription ids by hand (see the old examples/stories/subscriptions/client.py for what that looked like). The driver folds all of it into structured concurrency:

• Entering waits for the acknowledgment (spec: the ack MUST be the stream's first message), so sub.honored — the filter subset the server agreed to deliver — is always populated, making the spec's "client SHOULD check the acknowledged filter" a plain attribute read. Pre-ack failures raise from enter (server rejection as MCPError, a stream that dies unacknowledged as SubscriptionLost); nothing degrades silently.
• Iteration yields typed events — the same four frozen dataclasses the server publishes, now defined once in mcp/shared/subscriptions.py and re-exported from the server module (no import breaks). Events are level triggers with no payload beyond identity, so unconsumed duplicates deduplicate and the backlog is bounded by the filter's width by construction — no cap, no overflow policy.
• The stream's two endings are control flow, mirroring the spec's clean-end-vs-abrupt-drop distinction: the server's empty result ends the async for; a drop raises SubscriptionLost. No replay, no automatic re-listen — a reconnecting client refetches, which the docs show as a four-line loop.
• Exiting the context ends the subscription with the transport's own spelling via the base branch's plumbing: the POST's response stream is aborted over streamable HTTP (that wire's cancellation signal), notifications/cancelled is sent on stream transports. The driver itself is transport-blind.
• Multiple subscriptions demultiplex independently by their verbatim subscription ids; tool calls run freely beside open streams — the docs cover the watcher-task pattern explicitly.
• Client.listen() on a pre-2026 connection raises a typed ListenNotSupportedError steering to the legacy verbs (TypeScript SDK parity), and subscribe_resource/unsubscribe_resource now carry deprecation warnings on both Client and ClientSession, with a migration-guide entry.

Session-level mechanics worth a reviewer's attention: acks are consumed only for ids in the driver's minted namespace, so raw escape-hatch listens (driving send_request directly) still observe their acks through message_handler; delivered events are handed to the consumer after the message_handler tee, so the caching layer's eviction always completes before an event-triggered refetch can run; and session teardown settles every open route, so a watcher task in a sibling task group gets SubscriptionLost instead of hanging when the client closes. One transport fix rides along: a per-request SSE stream that drops without ever carrying an event id now resolves its request with CONNECTION_CLOSED instead of leaving the waiter pending for the session's lifetime.

How Has This Been Tested?

• Full suite: 100.00% line+branch coverage, pyright clean.
• 24 driver tests (in-process, public API): ack surfacing, all four event kinds, dedupe, graceful/lost/local endings, era steer, rejection-from-enter, degenerate immediate-result open, server-sent cancel pre- and post-ack, slot release on exit, concurrent demux, tee ordering, ack-timeout, read-timeout exemption, duplicate acks, teardown-unblocks-sibling-consumer, raw-id collision containment, retained-handle staleness, bare-string filter rejection.
• 4 interaction tests across the connect matrix (in-memory + streamable HTTP at 2026; the era guard on the 2025 cells) with requirement-manifest rows.
• Docs: the subscriptions page gained a client section backed by a tested snippet; the subscriptions story is rewritten from ~95 lines of hand-rolled demux to the driver and runs across its matrix.
• Manually against a real uvicorn server through the public Client: honored filter round-trip, both event kinds, exact-URI filter silence, exit-frees-slot proven by a second listen on the same session, refetch-after-change, and the legacy steer.

Breaking Changes

None. The event types moved to mcp/shared/subscriptions.py but remain importable from mcp.server.subscriptions; the deprecated subscription verbs keep working against 2025-era servers.

Types of changes

• Bug fix (non-breaking change which fixes an issue)
• New feature (non-breaking change which adds functionality)
• Breaking change (fix or feature that would cause existing functionality to change)
• Documentation update

Checklist

• I have read the MCP Documentation
• My code follows the repository's style guidelines
• New and existing tests pass locally
• I have added appropriate error handling
• I have added or updated documentation as needed

Additional context

The conformance suite has no client-role listen scenarios — the upstream SEP traceability file explicitly excludes the two client-side obligations (subscription-id correlation is client-internal demux, not wire-observable; the ack check has no wire-observable definition) and the stdio client harness is tracked separately upstream. Coverage here is therefore the interaction suite plus TypeScript-SDK behavioral parity; flagging per the repo's conformance rule so the exclusion is a deliberate acceptance, not an oversight.

Deliberate omissions, for the record: no callbacks layer (the TypeScript SDK routes listen events into its notification-handler registrations; here message_handler still sees every teed frame, so a handler-style consumer remains constructible), no auto-open-at-connect, no auto-re-listen, and no close() method — the context manager is the lifecycle.

[github-actions]

📚 Documentation preview

Preview	https://pr-3047.mcp-python-docs.pages.dev
Deployment	https://1e4c57c6.mcp-python-docs.pages.dev
Commit	4c95899
Triggered by	@maxisbey
Updated	2026-07-01 20:35:30 UTC

[cubic-dev-ai]

3 issues found across 23 files

<sub>Reply with feedback, questions, or to request a fix.

Fix all with cubic | Re-trigger cubic</sub>

[cubic-dev-ai]

2 issues found across 16 files (changes from recent commits).

Prompt for AI agents (unresolved issues)

Check if these issues are valid — if so, understand the root cause of each and fix them. If appropriate, use sub-agents to investigate and fix each issue separately.


<file name="docs/advanced/subscriptions.md">

<violation number="1" location="docs/advanced/subscriptions.md:91">
P2: The client docs now give conflicting filter semantics: this line says a subscription may receive sub-resource URIs, while the same page says resource subscriptions match exact URI strings. Consider keeping the event description aligned with exact-match delivery so clients do not build watchers that expect broader notifications.</violation>
</file>

<file name="src/mcp/client/subscriptions.py">

<violation number="1" location="src/mcp/client/subscriptions.py:153">
P2: A subscription to one resource URI currently accepts `ResourceUpdated` events for any URI, not just acknowledged ones. In `ListenRoute.deliver`, `ResourceUpdated` admission is `bool(self._honored_uris)`, so once at least one URI is honored, unrelated updates are queued and can wake consumers or even drive backlog-loss on noise. Aligning this branch with URI-based matching (same predicate used server-side) would keep delivery scoped to the acknowledged filter.</violation>
</file>

<sub>Reply with feedback, questions, or to request a fix.

Fix all with cubic | Re-trigger cubic</sub>

[cubic-dev-ai]

P2: The client docs now give conflicting filter semantics: this line says a subscription may receive sub-resource URIs, while the same page says resource subscriptions match exact URI strings. Consider keeping the event description aligned with exact-match delivery so clients do not build watchers that expect broader notifications.

Prompt for AI agents

Check if this issue is valid — if so, understand the root cause and fix it. At docs/advanced/subscriptions.md, line 91:

<comment>The client docs now give conflicting filter semantics: this line says a subscription may receive sub-resource URIs, while the same page says resource subscriptions match exact URI strings. Consider keeping the event description aligned with exact-match delivery so clients do not build watchers that expect broader notifications.</comment>

<file context>
@@ -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:
</file context>

Suggested change

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

[cubic-dev-ai]

P2: A subscription to one resource URI currently accepts ResourceUpdated events for any URI, not just acknowledged ones. In ListenRoute.deliver, ResourceUpdated admission is bool(self._honored_uris), so once at least one URI is honored, unrelated updates are queued and can wake consumers or even drive backlog-loss on noise. Aligning this branch with URI-based matching (same predicate used server-side) would keep delivery scoped to the acknowledged filter.

Prompt for AI agents

Check if this issue is valid — if so, understand the root cause and fix it. At src/mcp/client/subscriptions.py, line 153:

<comment>A subscription to one resource URI currently accepts `ResourceUpdated` events for any URI, not just acknowledged ones. In `ListenRoute.deliver`, `ResourceUpdated` admission is `bool(self._honored_uris)`, so once at least one URI is honored, unrelated updates are queued and can wake consumers or even drive backlog-loss on noise. Aligning this branch with URI-based matching (same predicate used server-side) would keep delivery scoped to the acknowledged filter.</comment>

<file context>
@@ -89,36 +101,71 @@ class SubscriptionLost(RuntimeError):
+        if self.end is not None or self.honored is None:
+            return
+        if isinstance(event, ResourceUpdated):
+            admitted = bool(self._honored_uris)
+        else:
+            admitted = event_matches(self.honored, self._honored_uris, event)
</file context>