Add the client-side subscriptions/listen driver

docs/client/index.md

@@ -145,7 +145,7 @@ The resource verbs come in pairs: two ways to list, one way to read.
 
 `read_resource` returns `contents`, a list of `TextResourceContents` or `BlobResourceContents`. Same idea as tool content: narrow with `isinstance`, then read `.text` (or `.blob`).
 
-A client can also be told when a resource changes. On 2025-era connections that is `subscribe_resource(uri)` / `unsubscribe_resource(uri)` - a method pair `MCPServer` doesn't implement, so on the 2026-07-28 wire (where those verbs no longer exist) the request answers `-32601`, *Method not found*. The 2026 replacement is a `subscriptions/listen` stream, which `MCPServer` *does* serve - `server_capabilities.resources.subscribe` is `True` there, and the server side of the story is **[Subscriptions](../handlers/subscriptions.md)**.
+A client can also be told when a resource changes. On 2025-era connections that is `subscribe_resource(uri)` / `unsubscribe_resource(uri)` - a method pair `MCPServer` doesn't implement, so on the 2026-07-28 wire (where those verbs no longer exist) the request answers `-32601`, *Method not found*. The 2026 replacement is a `subscriptions/listen` stream, which `MCPServer` *does* serve - `server_capabilities.resources.subscribe` is `True` there, and **[Subscriptions](../handlers/subscriptions.md)** tells both sides of the story - the client end is [`client.listen(...)`](../handlers/subscriptions.md#the-client-side).
 
 ## Prompts
 

docs/handlers/subscriptions.md

@@ -85,10 +85,32 @@ Down on the low-level `Server` there is no pre-wired anything — and the same p
 * `ListenHandler(bus)` is the same handler `MCPServer` registers; `on_subscriptions_listen=` is an ordinary handler slot. Don't want the SDK's semantics? Write your own handler for the slot — the spec obligations come with it.
 * `ListenHandler.close()` gracefully ends every open stream: each one receives the listen request's result as its final frame, the spec's signal that the server ended the subscription deliberately — a clean end, as opposed to the abrupt drop a client may treat as a cue to reconnect. Without it, streams end when the client disconnects.
 
+## The client side
+
+Consuming a subscription is one context manager:
+
+```python title="client.py" hl_lines="9 10"
+--8<-- "docs_src/subscriptions/tutorial003.py"
+```
+
+* `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=...)` — 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`. 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 title="watch.py" hl_lines="15 16"
+--8<-- "docs_src/subscriptions/tutorial004.py"
+```
+
+* 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.
+
 ## Recap
 
 * A client opts in with one `subscriptions/listen` request; the response is the stream. There is nothing to configure server-side — serving it is built in.
 * You publish: `await ctx.notify_resource_updated(uri)`, `notify_tools_changed()`, `notify_prompts_changed()`, `notify_resources_changed()`. Idle servers make these free.
 * Streams receive only what their filter requested; URIs match exactly; nothing is replayed.
 * Scaling out means implementing `SubscriptionBus` — two methods — over your own pub/sub, and passing it as `MCPServer(subscriptions=...)`.
 * The low-level spelling is the same machinery held in your hands: a bus, `ListenHandler(bus)`, one constructor argument.
+* Consuming is `async with client.listen(...)` and `async for event in sub` — typed events, honored filter on the handle, clean end vs `SubscriptionLost`.

docs/migration.md

@@ -1536,6 +1536,23 @@ The 2026-07-28 revision reintroduces Tasks as an official extension: [SEP-2663](
 
 ## Deprecations
 
+### Client resource-subscription methods deprecated (SEP-2575)
+
+[SEP-2575](https://github.com/modelcontextprotocol/modelcontextprotocol/issues/2575) removes `resources/subscribe` and `resources/unsubscribe` from the 2026-07-28 wire; per-URI subscriptions travel in the `subscriptions/listen` filter instead. The client verbs now carry `typing_extensions.deprecated`:
+
+- `Client.subscribe_resource()` / `Client.unsubscribe_resource()`
+- `ClientSession.subscribe_resource()` / `ClientSession.unsubscribe_resource()`
+
+They keep working against 2025-era servers; a 2026-07-28 server answers them with `-32601` (method not found). Migrate to the listen driver:
+
+```python
+async with client.listen(resource_subscriptions=["note://todo"]) as sub:
+    async for event in sub:  # ResourceUpdated(uri="note://todo")
+        ...
+```
+
+See the [Subscriptions](handlers/subscriptions.md#the-client-side) page for the full client-side contract (typed events, the honored filter, clean end vs `SubscriptionLost`).
+
 ### Roots, Sampling, and Logging methods deprecated (SEP-2577)
 
 [SEP-2577](https://github.com/modelcontextprotocol/modelcontextprotocol/pull/2577) deprecates the Roots, Sampling, and Logging features as of the 2026-07-28 spec. The deprecation is advisory only: there are no wire-level changes, capability negotiation is unchanged, and every method keeps working for sessions negotiating 2025-11-25 and earlier.

docs/whats-new.md

@@ -190,9 +190,9 @@ 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`. Two honest caveats as of `2.0.0b1`: the Python `Client` cannot open the listen stream yet (the driver ships in a later pre-release), and over stdio the server does not serve it. The net for a Python *client* on that release is that nothing delivers change notifications on a 2026-07-28 connection; a host that relies on `resources/updated` should connect with `mode="legacy"` until the driver lands.
+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.
 
-**[Subscriptions](handlers/subscriptions.md)** on the server, and **[Deploy & scale](run/deploy.md)** for the bus.
+**[Subscriptions](handlers/subscriptions.md)** covers both sides, and **[Deploy & scale](run/deploy.md)** the bus.
 
 ### The rest, quickly
 

docs_src/subscriptions/tutorial003.py

@@ -0,0 +1,14 @@
+from mcp import Client
+from mcp.client.subscriptions import ResourceUpdated
+
+from .tutorial001 import mcp
+
+
+async def watch_todo() -> str:
+    """Wait for the todo note to change once, then stop listening."""
+    async with Client(mcp) as client:
+        async with client.listen(resource_subscriptions=["note://todo"]) as sub:
+            async for event in sub:
+                assert isinstance(event, ResourceUpdated)
+                return f"changed: {event.uri}"
+    return "the server closed the stream before any change"

docs_src/subscriptions/tutorial004.py

@@ -0,0 +1,20 @@
+import anyio
+
+from mcp import Client
+from mcp.client.subscriptions import SubscriptionLost
+
+
+async def watch(client: Client, uri: str) -> None:
+    """Keep one resource fresh for as long as the client lives."""
+    while True:
+        try:
+            async with client.listen(resource_subscriptions=[uri]) as sub:
+                await client.read_resource(uri)  # refetch: no replay across streams
+                async for _event in sub:
+                    await client.read_resource(uri)
+        except SubscriptionLost:
+            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)

examples/stories/subscriptions/client.py

@@ -4,88 +4,34 @@
 import mcp_types as types
 
 from mcp.client import Client
+from mcp.client.subscriptions import ResourceUpdated, ToolsListChanged
 from stories._harness import Target, run_client
 
-SUBSCRIPTION_ID = "io.modelcontextprotocol/subscriptionId"
-
 
 async def main(target: Target, *, mode: str = "auto") -> None:
-    # Stream frames arrive as ordinary server notifications; `message_handler`
-    # is constructor-only on `Client`, so the list it fills exists first.
-    received: list[types.ServerNotification] = []
-    arrival = anyio.Event()
-
-    async def on_message(message: object) -> None:
-        nonlocal arrival
-        if isinstance(
-            message,
-            types.SubscriptionsAcknowledgedNotification
-            | types.ResourceUpdatedNotification
-            | types.ToolListChangedNotification,
-        ):
-            received.append(message)
-            arrival.set()
-            arrival = anyio.Event()
-
-    async def wait_for(count: int) -> None:
-        with anyio.fail_after(10):
-            while len(received) < count:
-                await arrival.wait()
-
-    async with Client(target, mode=mode, message_handler=on_message) as client:
+    async with Client(target, mode=mode) as client:
         before = await client.list_tools()
         assert "search" not in {tool.name for tool in before.tools}
 
-        async with anyio.create_task_group() as tg:
-            # There is no client-side listen API yet, so the story drops to the
-            # `client.session` escape hatch. The request parks for the stream's
-            # lifetime, so it runs as a task; cancelling it releases the local
-            # awaiting scope. In-memory that also ends the server's stream; over
-            # HTTP today nothing aborts the POST, so the server-side stream ends
-            # when the connection closes (the `Client` exit right below).
-            async def listen() -> None:
-                request = types.SubscriptionsListenRequest(
-                    params=types.SubscriptionsListenRequestParams(
-                        notifications=types.SubscriptionFilter(
-                            tools_list_changed=True, resource_subscriptions=["note://todo"]
-                        )
-                    )
-                )
-                await client.session.send_request(request, types.SubscriptionsListenResult)
-
-            tg.start_soon(listen)
-
-            # ── the ack is the first frame: it echoes the honored filter, tagged ──
-            await wait_for(1)
-            ack = received[0]
-            assert isinstance(ack, types.SubscriptionsAcknowledgedNotification), ack
-            assert ack.params.notifications.tools_list_changed is True
-            assert ack.params.notifications.resource_subscriptions == ["note://todo"]
-            assert ack.params.meta is not None and SUBSCRIPTION_ID in ack.params.meta
+        async with client.listen(tools_list_changed=True, resource_subscriptions=["note://todo"]) as sub:
+            # ── entering waited for the ack: the honored filter is already in hand ──
+            assert sub.honored.tools_list_changed is True
+            assert sub.honored.resource_subscriptions == ["note://todo"]
 
             # ── exact-URI filtering: an unsubscribed note edit stays silent ──
             await client.call_tool("edit_note", {"name": "journal", "text": "day two"})
-            # ── the subscribed URI delivers, carrying the same subscription id ──
+            # ── the subscribed URI delivers ──
             await client.call_tool("edit_note", {"name": "todo", "text": "water plants"})
-            await wait_for(2)
-            updated = received[1]
-            assert isinstance(updated, types.ResourceUpdatedNotification), updated
-            assert updated.params.uri == "note://todo"
-            assert updated.params.meta is not None
-            assert updated.params.meta[SUBSCRIPTION_ID] == ack.params.meta[SUBSCRIPTION_ID]
-            assert len(received) == 2, "the journal edit must not have been delivered"
+            with anyio.fail_after(10):
+                event = await anext(sub)
+            assert event == ResourceUpdated(uri="note://todo"), "the journal edit must not have been delivered"
 
             # ── a runtime tool registration announces itself ──
             await client.call_tool("enable_search", {})
-            await wait_for(3)
-            assert isinstance(received[2], types.ToolListChangedNotification), received[2]
-
-            # The client is done listening: cancel the parked request and let
-            # the connection teardown below end the stream server-side.
-            tg.cancel_scope.cancel()
+            with anyio.fail_after(10):
+                assert await anext(sub) == ToolsListChanged()
 
-        # list_changed told us to re-fetch - the new tool is callable, and the
-        # session outlives the closed stream.
+        # ── leaving the block closed the stream; the session lives on ──
         tools = await client.list_tools()
         assert "search" in {tool.name for tool in tools.tools}
         result = await client.call_tool("search", {"query": "water"})