Add the client subscriptions/listen surface

Client.listen(filter) opens a subscription as an async context manager:
it registers demux state before sending, resolves only on the server's
acknowledgment (raising on error responses, transport loss, or timeout),
surfaces the honored and narrowed filter subsets, and yields typed
change notifications via async iteration. Graceful server closure and
local close end iteration cleanly; an abrupt transport drop raises
SubscriptionLost so a dead stream cannot be missed silently. strict=True
turns acknowledgment narrowing into ListenNarrowedError.

Dispatchers accept a caller-supplied request id (CallOptions request_id)
so the subscription id is known before the request is sent; client-
minted string ids stay collision-free against the numeric counter by
type. ClientSession routes the acknowledgment and tagged stream
notifications to the live subscription by the verbatim-typed
subscription id and tees matched notifications onward, so message
handlers and response-cache eviction keep working unchanged.

On 2026 streamable HTTP, closing a subscription aborts the listen
request's response stream: the transport translates the outbound
cancellation instead of posting it, since stream closure is the
protocol's cancellation signal. The listen request is exempt from read
timeouts; only the acknowledgment wait is bounded.

subscribe_resource and unsubscribe_resource are deprecated with
pointers to Client.listen, and the migration guide documents the
replacement.

Author: maxisbey

docs/migration.md

@@ -1517,6 +1517,10 @@ The 2026-07-28 spec restricts `notifications/progress` to the server-to-client d
 
 On the server side, prefer the new dispatcher-agnostic `ServerSession.report_progress(progress, total, message)` (and `Context.report_progress()` on `MCPServer`) over the raw `ServerSession.send_progress_notification(progress_token, …)`. `report_progress` encapsulates the "no-op when the caller did not request progress" rule and works on every dispatcher; the raw token-taking form remains for handlers that read `_meta.progressToken` directly.
 
+### Client resource-subscription methods deprecated (2026-07-28)
+
+The 2026-07-28 wire removes `resources/subscribe` and `resources/unsubscribe`: resource subscriptions are delivered over the `subscriptions/listen` stream instead (see [`MCPServer` serves `subscriptions/listen`](#mcpserver-serves-subscriptionslisten-json-response-mode-requires-subscriptionsfalse-sep-2575)). `Client.subscribe_resource()` / `Client.unsubscribe_resource()` and the `ClientSession` pair now carry `typing_extensions.deprecated` and emit `mcp.MCPDeprecationWarning` at runtime. They keep working against servers negotiating 2025-11-25 or earlier; on a 2026-07-28 connection the server answers them with `-32601` (Method not found). New code should open a subscription stream with `client.listen(...)`.
+
 ## Bug Fixes
 
 ### OAuth metadata URLs no longer gain a trailing slash

src/mcp/client/client.py

@@ -5,16 +5,18 @@
 import hashlib
 import logging
 import uuid
-from collections.abc import Awaitable, Callable, Mapping
-from contextlib import AsyncExitStack
+from collections.abc import AsyncIterator, Awaitable, Callable, Iterator, Mapping
+from contextlib import AsyncExitStack, asynccontextmanager
 from dataclasses import KW_ONLY, dataclass, field
+from itertools import count
 from typing import Any, Literal, TypeVar, cast
 
 import anyio
 import anyio.lowlevel
 import mcp_types as types
 from mcp_types import (
     INVALID_PARAMS,
+    SUBSCRIPTION_ID_META_KEY,
     CacheableResult,
     CallToolResult,
     CompleteResult,
@@ -37,6 +39,7 @@
     RequestParamsMeta,
     ResourceTemplateReference,
     ServerCapabilities,
+    SubscriptionFilter,
 )
 from mcp_types.version import HANDSHAKE_PROTOCOL_VERSIONS, MODERN_PROTOCOL_VERSIONS
 from typing_extensions import deprecated
@@ -56,11 +59,12 @@
     SamplingFnT,
 )
 from mcp.client.streamable_http import streamable_http_client
+from mcp.client.subscriptions import ListenNarrowedError, ListenNotSupportedError, Subscription
 from mcp.server import Server
 from mcp.server.mcpserver import MCPServer
 from mcp.server.runner import modern_on_request
 from mcp.shared.direct_dispatcher import create_direct_dispatcher_pair
-from mcp.shared.dispatcher import Dispatcher, ProgressFnT
+from mcp.shared.dispatcher import CallOptions, Dispatcher, ProgressFnT
 from mcp.shared.exceptions import MCPDeprecationWarning, MCPError
 from mcp.shared.jsonrpc_dispatcher import JSONRPCDispatcher
 from mcp.shared.session import RequestResponder
@@ -282,6 +286,7 @@ async def main():
     `target_id` when the server is not a URL (no identity can be derived)."""
 
     _entered: bool = field(init=False, default=False)
+    _listen_ids: Iterator[int] = field(init=False, default_factory=lambda: count(1), repr=False, compare=False)
     _session: ClientSession | None = field(init=False, default=None)
     _exit_stack: AsyncExitStack | None = field(init=False, default=None)
     _connect: _Connector = field(init=False, repr=False, compare=False)
@@ -583,13 +588,139 @@ async def retry(r: InputResponses | None, s: str | None) -> ReadResourceResult |
         # Driver rounds carry inputResponses, so a terminal result reached through them is never cached (spec MUST).
         return await self._drive_input_required(first, retry)
 
+    @deprecated(
+        "resources/subscribe is removed as of 2026-07-28; use Client.listen() instead.",
+        category=MCPDeprecationWarning,
+    )
     async def subscribe_resource(self, uri: str, *, meta: RequestParamsMeta | None = None) -> EmptyResult:
         """Subscribe to resource updates."""
-        return await self.session.subscribe_resource(uri, meta=meta)
+        return await self.session.subscribe_resource(uri, meta=meta)  # pyright: ignore[reportDeprecated]
 
+    @deprecated(
+        "resources/unsubscribe is removed as of 2026-07-28; use Client.listen() instead.",
+        category=MCPDeprecationWarning,
+    )
     async def unsubscribe_resource(self, uri: str, *, meta: RequestParamsMeta | None = None) -> EmptyResult:
         """Unsubscribe from resource updates."""
-        return await self.session.unsubscribe_resource(uri, meta=meta)
+        return await self.session.unsubscribe_resource(uri, meta=meta)  # pyright: ignore[reportDeprecated]
+
+    @asynccontextmanager
+    async def listen(
+        self,
+        notifications: SubscriptionFilter,
+        *,
+        strict: bool = False,
+        ack_timeout: float = 30.0,
+    ) -> AsyncIterator[Subscription]:
+        """Open a `subscriptions/listen` stream (2026-07-28) and yield its `Subscription`.
+
+        Entering the context sends the listen request and returns once the
+        server's acknowledgement arrives; iterate the yielded `Subscription`
+        for the stream's typed change notifications. Exiting the context
+        closes the stream. The listen request itself is exempt from
+        `read_timeout_seconds` — only the acknowledgement phase is bounded
+        (`ack_timeout`). A server result before any acknowledgement is
+        tolerated as a degenerate immediately-graceful open: nothing honored,
+        iteration ends at once.
+
+        Overlapping streams are independent handles: a notification matching
+        several open filters is delivered to each of them.
+
+        Args:
+            notifications: The notification types to opt in to.
+            strict: Raise `ListenNarrowedError` (closing the stream) when the
+                server honors less than requested, instead of returning the
+                narrowed subscription.
+            ack_timeout: Seconds to wait for the server's acknowledgement.
+
+        Raises:
+            ListenNotSupportedError: The negotiated protocol version predates
+                `subscriptions/listen`.
+            ListenNarrowedError: `strict=True` and the server narrowed the filter.
+            MCPError: The server rejected the request, or the connection
+                closed before the acknowledgement.
+            TimeoutError: No acknowledgement within `ack_timeout`.
+        """
+        session = self.session
+        if session.protocol_version not in MODERN_PROTOCOL_VERSIONS:
+            raise ListenNotSupportedError(session.protocol_version)
+        subscription_id = f"listen-{next(self._listen_ids)}"
+        request = types.SubscriptionsListenRequest(
+            params=types.SubscriptionsListenRequestParams(notifications=notifications)
+        )
+        data = request.model_dump(by_alias=True, mode="json", exclude_none=True)
+        # The session's stamp (modern `_meta` envelope + transport headers) but NOT
+        # its read-timeout fallback: a listen stream must outlive any session
+        # default, so the dispatcher is driven directly, with no timeout at all.
+        opts: CallOptions = {"request_id": subscription_id}
+        session._stamp(data, opts)  # pyright: ignore[reportPrivateUsage]
+        # The modern stamp suppresses the courtesy cancel, but `close()` relies on
+        # it: at a modern version the frame never crosses the wire - the transport
+        # translates it into aborting the listen POST's response stream (LR-15).
+        opts["cancel_on_abandon"] = True
+        open_error: MCPError | None = None
+        driver_scope = anyio.CancelScope()
+        # Registered BEFORE the send, so the acknowledgement cannot race the route.
+        route = session._register_listen_route(subscription_id)  # pyright: ignore[reportPrivateUsage]
+
+        async def drive() -> None:
+            nonlocal open_error
+            with driver_scope:
+                try:
+                    raw = await session._dispatcher.send_raw_request(  # pyright: ignore[reportPrivateUsage]
+                        data["method"], data.get("params"), opts
+                    )
+                except MCPError as exc:
+                    # Error response or connection close: pre-ack this wakes
+                    # `__aenter__` immediately (the settle sets the ack event);
+                    # post-ack, iteration raises `SubscriptionLost`.
+                    open_error = exc
+                    route.settle("remote")
+                    return
+                # Receipt of the result IS the graceful close (validation is
+                # tolerant: a foreign shape is logged, never an error).
+                raw_meta = raw.get("_meta")
+                meta = cast("dict[str, Any]", raw_meta) if isinstance(raw_meta, dict) else {}
+                if meta.get(SUBSCRIPTION_ID_META_KEY) != subscription_id:
+                    logger.debug("subscriptions/listen result for %r missing its subscription-id meta", subscription_id)
+                # A result before any ack is the degenerate immediately-graceful
+                # open: nothing honored. After a real ack this is a no-op.
+                route.set_acked(SubscriptionFilter())
+                route.settle("graceful")
+
+        # The driver rides the session's own task group: a local task group here
+        # would wrap every exception crossing the `yield` — the typed open
+        # failures below AND the caller's own errors — in an ExceptionGroup.
+        task_group = session._task_group  # pyright: ignore[reportPrivateUsage]
+        assert task_group is not None  # an entered session (the `.session` gate) always has one
+        try:
+            task_group.start_soon(drive)
+            try:
+                with anyio.fail_after(ack_timeout):
+                    await route.acked.wait()
+            except TimeoutError:
+                raise TimeoutError(f"server did not acknowledge subscriptions/listen within {ack_timeout}s") from None
+            if open_error is not None and route.honored is None:
+                raise open_error
+            subscription = Subscription(
+                subscription_id=subscription_id,
+                requested=notifications,
+                # None only when the route settled remotely before any ack:
+                # the handle then surfaces it as `SubscriptionLost` on iteration.
+                honored=route.honored if route.honored is not None else SubscriptionFilter(),
+                route=route,
+                cancel_driver=driver_scope.cancel,
+            )
+            if strict and subscription.narrowed != SubscriptionFilter():
+                raise ListenNarrowedError(subscription.narrowed, subscription.honored)
+            yield subscription
+        finally:
+            # One teardown for every path (failed open, close(), plain exit):
+            # the same settle-then-cancel `Subscription.close()` performs, all
+            # idempotent against whatever reason already won.
+            route.settle("local")
+            driver_scope.cancel()
+            session._unregister_listen_route(subscription_id)  # pyright: ignore[reportPrivateUsage]
 
     async def call_tool(
         self,