Add the client-side subscriptions/listen driver (#3047)

tests/client/test_client.py

@@ -35,7 +35,7 @@ from mcp_types import (
 from mcp_types.version import LATEST_HANDSHAKE_VERSION
 from pydantic import FileUrl
 
-from mcp import MCPError
+from mcp import MCPDeprecationWarning, MCPError
 from mcp.client._memory import InMemoryTransport
 from mcp.client._transport import TransportStreams
 from mcp.client.client import Client
@@ -310,13 +310,15 @@ async def test_client_send_progress_notification():
 
 async def test_client_subscribe_resource(simple_server: Server):
     async with Client(simple_server, mode="legacy") as client:
-        result = await client.subscribe_resource("memory://test")
+        with pytest.warns(MCPDeprecationWarning, match="use Client.listen"):
+            result = await client.subscribe_resource("memory://test")  # pyright: ignore[reportDeprecated]
         assert result == snapshot(EmptyResult())
 
 
 async def test_client_unsubscribe_resource(simple_server: Server):
     async with Client(simple_server, mode="legacy") as client:
-        result = await client.unsubscribe_resource("memory://test")
+        with pytest.warns(MCPDeprecationWarning, match="use Client.listen"):
+            result = await client.unsubscribe_resource("memory://test")  # pyright: ignore[reportDeprecated]
         assert result == snapshot(EmptyResult())
 
 

tests/client/test_send_request_mcp_name.py

@@ -22,7 +22,7 @@ from mcp_types import (
 from mcp_types.version import LATEST_HANDSHAKE_VERSION, LATEST_MODERN_VERSION
 
 from mcp.client.session import ClientSession
-from mcp.shared.dispatcher import CallOptions, OnNotify, OnRequest
+from mcp.shared.dispatcher import CallOptions, OnNotify, OnNotifyIntercept, OnRequest
 from mcp.shared.inbound import MCP_NAME_HEADER, MCP_PROTOCOL_VERSION_HEADER, encode_header_value
 
 
@@ -36,6 +36,7 @@ class _RecordingDispatcher:
         self,
         on_request: OnRequest,
         on_notify: OnNotify,
+        on_notify_intercept: OnNotifyIntercept | None = None,
         *,
         task_status: anyio.abc.TaskStatus[None] = anyio.TASK_STATUS_IGNORED,
     ) -> None:

tests/client/test_session.py

@@ -39,11 +39,13 @@ from mcp import MCPError
 from mcp.client import ClientRequestContext
 from mcp.client.client import Client
 from mcp.client.session import DEFAULT_CLIENT_INFO, ClientSession
+from mcp.client.subscriptions import ToolsListChanged, listen
 from mcp.server import Server, ServerRequestContext
 from mcp.shared.direct_dispatcher import create_direct_dispatcher_pair
-from mcp.shared.dispatcher import CallOptions, DispatchContext, OnNotify, OnRequest
+from mcp.shared.dispatcher import CallOptions, DispatchContext, OnNotify, OnNotifyIntercept, OnRequest
 from mcp.shared.message import SessionMessage
 from mcp.shared.session import RequestResponder
+from mcp.shared.subscriptions import SUBSCRIPTION_ID_META_KEY
 from mcp.shared.transport_context import TransportContext
 
 _SendToClient = anyio.streams.memory.MemoryObjectSendStream[SessionMessage | Exception]
@@ -1341,6 +1343,7 @@ class _OptsRecordingDispatcher:
         self,
         on_request: OnRequest,
         on_notify: OnNotify,
+        on_notify_intercept: OnNotifyIntercept | None = None,
         *,
         task_status: anyio.abc.TaskStatus[None] = anyio.TASK_STATUS_IGNORED,
     ) -> None:
@@ -1428,6 +1431,7 @@ async def test_aenter_cancelled_while_dispatcher_starts_unwinds_cleanly():
             self,
             on_request: OnRequest,
             on_notify: OnNotify,
+            on_notify_intercept: OnNotifyIntercept | None = None,
             *,
             task_status: anyio.abc.TaskStatus[None] = anyio.TASK_STATUS_IGNORED,
         ) -> None:
@@ -1486,6 +1490,7 @@ class _ScriptedDispatcher:
         self,
         on_request: OnRequest,
         on_notify: OnNotify,
+        on_notify_intercept: OnNotifyIntercept | None = None,
         *,
         task_status: anyio.abc.TaskStatus[None] = anyio.TASK_STATUS_IGNORED,
     ) -> None:
@@ -1808,3 +1813,137 @@ async def test_session_read_resource_returns_input_required_result_when_opted_in
             result = await client.session.read_resource("memory://r", allow_input_required=True)
     assert isinstance(result, types.InputRequiredResult)
     assert result.request_state == "resource-state"
+
+
+@pytest.mark.anyio
+async def test_a_late_ack_for_a_closed_driver_listen_reaches_message_handler():
+    """Ack consumption is keyed on the live route registry alone: a stray ack for a
+    closed subscription's id surfaces through message_handler like any other unowned frame."""
+    seen: list[object] = []
+    follow_up = anyio.Event()
+
+    async def handler(msg: object) -> None:
+        seen.append(msg)
+        if len(seen) == 2:
+            follow_up.set()
+
+    async with raw_client_session(message_handler=handler) as (session, to_client, _):
+        _set_negotiated_version(session, "2026-07-28")
+        session._register_listen_route("listen-99")  # pyright: ignore[reportPrivateUsage]
+        session._unregister_listen_route("listen-99")  # pyright: ignore[reportPrivateUsage]
+        await to_client.send(
+            SessionMessage(
+                JSONRPCNotification(
+                    jsonrpc="2.0",
+                    method="notifications/subscriptions/acknowledged",
+                    params={
+                        "notifications": {"toolsListChanged": True},
+                        "_meta": {SUBSCRIPTION_ID_META_KEY: "listen-99"},
+                    },
+                )
+            )
+        )
+        await to_client.send(
+            SessionMessage(JSONRPCNotification(jsonrpc="2.0", method="notifications/tools/list_changed", params={}))
+        )
+        with anyio.fail_after(5):
+            await follow_up.wait()
+    assert [type(message).__name__ for message in seen] == [
+        "SubscriptionsAcknowledgedNotification",
+        "ToolListChangedNotification",
+    ]
+
+
+@pytest.mark.anyio
+async def test_a_graceful_result_does_not_outrun_the_events_that_preceded_it():
+    """[ack, event, result] written back-to-back: the event delivers and the wire ack's filter
+    survives a parked message_handler tee, because routes settle on the dispatcher's receive path in wire order."""
+
+    async def parked_handler(message: object) -> None:
+        await anyio.sleep_forever()
+
+    events: list[object] = []
+    honored: list[types.SubscriptionFilter] = []
+    async with raw_client_session(message_handler=parked_handler) as (session, to_client, from_client):
+        _set_negotiated_version(session, "2026-07-28")
+        with anyio.fail_after(5):
+            async with anyio.create_task_group() as tg:  # pragma: no branch
+
+                async def consume() -> None:
+                    async with listen(session, tools_list_changed=True) as sub:  # pragma: no branch
+                        honored.append(sub.honored)
+                        events.extend([event async for event in sub])
+
+                tg.start_soon(consume)
+                request = await from_client.receive()
+                assert isinstance(request.message, JSONRPCRequest)
+                meta = {SUBSCRIPTION_ID_META_KEY: request.message.id}
+                for message in (
+                    JSONRPCNotification(
+                        jsonrpc="2.0",
+                        method="notifications/subscriptions/acknowledged",
+                        params={"notifications": {"toolsListChanged": True}, "_meta": meta},
+                    ),
+                    JSONRPCNotification(
+                        jsonrpc="2.0", method="notifications/tools/list_changed", params={"_meta": meta}
+                    ),
+                    JSONRPCResponse(jsonrpc="2.0", id=request.message.id, result={"_meta": meta}),
+                ):
+                    await to_client.send(SessionMessage(message))
+    assert honored == [types.SubscriptionFilter(tools_list_changed=True)]
+    assert events == [ToolsListChanged()]
+
+
+def _intercept_only_session() -> ClientSession:
+    """A never-entered session whose intercept can be driven directly (it is synchronous)."""
+    dispatcher, _peer = create_direct_dispatcher_pair()
+    return ClientSession(dispatcher=dispatcher)
+
+
+def test_intercept_settles_only_the_named_listen_route_on_cancelled():
+    """SDK demux contract: a server-sent cancel settles exactly the listen route it names and is never consumed."""
+    session = _intercept_only_session()
+    route = session._register_listen_route("listen-1")  # pyright: ignore[reportPrivateUsage]
+    intercept = session._intercept_notification  # pyright: ignore[reportPrivateUsage]
+    assert intercept("notifications/cancelled", {"requestId": "unrelated"}) is False
+    assert route.end is None
+    assert intercept("notifications/cancelled", {"requestId": "listen-1"}) is False
+    assert route.end == "lost"
+
+
+def test_intercept_ignores_frames_without_a_route_or_with_broken_meta():
+    """SDK demux contract: frames that correlate to no live route flow through to the normal notification path."""
+    session = _intercept_only_session()
+    intercept = session._intercept_notification  # pyright: ignore[reportPrivateUsage]
+    assert intercept("notifications/tools/list_changed", {"_meta": {SUBSCRIPTION_ID_META_KEY: "listen-1"}}) is False
+    route = session._register_listen_route("listen-1")  # pyright: ignore[reportPrivateUsage]
+    route.set_acked(types.SubscriptionFilter(tools_list_changed=True))
+    assert intercept("notifications/tools/list_changed", None) is False
+    # A non-mapping `_meta` is constructible on pre-2026 wires.
+    assert intercept("notifications/tools/list_changed", {"_meta": "oops"}) is False
+    assert intercept("notifications/tools/list_changed", {"_meta": {SUBSCRIPTION_ID_META_KEY: "other"}}) is False
+    # A non-string uri is not an event; surface validation owns it.
+    meta = {"_meta": {SUBSCRIPTION_ID_META_KEY: "listen-1"}}
+    assert intercept("notifications/resources/updated", {"uri": 7, **meta}) is False
+    assert route._pending == {}  # pyright: ignore[reportPrivateUsage]
+
+
+def test_intercept_consumes_acks_for_live_routes_and_leaves_malformed_ones():
+    """SDK demux contract: a well-formed ack for a live route is consumed as driver state; malformed acks pass on."""
+    session = _intercept_only_session()
+    route = session._register_listen_route("listen-1")  # pyright: ignore[reportPrivateUsage]
+    intercept = session._intercept_notification  # pyright: ignore[reportPrivateUsage]
+    meta = {"_meta": {SUBSCRIPTION_ID_META_KEY: "listen-1"}}
+    assert intercept("notifications/subscriptions/acknowledged", {"notifications": ["nope"], **meta}) is False
+    assert route.honored is None
+    # A missing `notifications` field must not be read as an (all-refusing) empty filter.
+    assert intercept("notifications/subscriptions/acknowledged", dict(meta)) is False
+    assert route.honored is None
+    assert (
+        intercept("notifications/subscriptions/acknowledged", {"notifications": {"toolsListChanged": True}, **meta})
+        is True
+    )
+    assert route.honored == types.SubscriptionFilter(tools_list_changed=True)
+    # Events deliver but are never consumed - they still tee to message_handler.
+    assert intercept("notifications/tools/list_changed", meta) is False
+    assert list(route._pending) == [ToolsListChanged()]  # pyright: ignore[reportPrivateUsage]

tests/client/test_session_claims.py

@@ -28,7 +28,7 @@ from typing_extensions import assert_type
 
 from mcp.client.extension import ClaimContext, ResultClaim, UnexpectedClaimedResult
 from mcp.client.session import ClientSession, _CallToolResultAdapter
-from mcp.shared.dispatcher import CallOptions, OnNotify, OnRequest
+from mcp.shared.dispatcher import CallOptions, OnNotify, OnNotifyIntercept, OnRequest
 
 _TASKS_EXT = "com.example/tasks"
 _AD_ONLY_EXT = "com.example/flags"
@@ -75,6 +75,7 @@ class _RecordingDispatcher:
         self,
         on_request: OnRequest,
         on_notify: OnNotify,
+        on_notify_intercept: OnNotifyIntercept | None = None,
         *,
         task_status: anyio.abc.TaskStatus[None] = anyio.TASK_STATUS_IGNORED,
     ) -> None:

tests/client/test_streamable_http.py

@@ -18,6 +18,8 @@ from inline_snapshot import snapshot
 from mcp_types import (
     CLIENT_CAPABILITIES_META_KEY,
     CLIENT_INFO_META_KEY,
+    CONNECTION_CLOSED,
+    INVALID_REQUEST,
     METHOD_NOT_FOUND,
     PROTOCOL_VERSION_META_KEY,
     JSONRPCError,
@@ -28,10 +30,16 @@ from mcp_types import (
 from mcp_types.version import LATEST_MODERN_VERSION
 from starlette.types import Receive, Scope, Send
 
-from mcp.client.streamable_http import streamable_http_client
+from mcp.client.streamable_http import (
+    MAX_RECONNECTION_ATTEMPTS,
+    RequestContext,
+    StreamableHTTPTransport,
+    streamable_http_client,
+)
 from mcp.server import Server
 from mcp.server._streamable_http_modern import handle_modern_request
 from mcp.server.subscriptions import InMemorySubscriptionBus, ListenHandler, ServerEvent
+from mcp.shared._context_streams import ContextSendStream, create_context_streams
 from mcp.shared.dispatcher import CallOptions, DispatchContext
 from mcp.shared.inbound import MCP_METHOD_HEADER, MCP_PROTOCOL_VERSION_HEADER, encode_header_value
 from mcp.shared.jsonrpc_dispatcher import JSONRPCDispatcher
@@ -583,3 +591,160 @@ async def test_a_finished_post_task_does_not_evict_a_reused_ids_new_registration
             )
             await parked.closed.wait()
     assert [body["method"] for body in posted] == ["tools/call", "subscriptions/listen"]
+
+
+class _DyingSSEStream(httpx.AsyncByteStream):
+    """Emits one id-less comment then breaks - a non-resumable stream dropping."""
+
+    def __init__(self) -> None:
+        self.opened = anyio.Event()
+
+    async def __aiter__(self) -> AsyncIterator[bytes]:
+        self.opened.set()
+        yield b": hello\n\n"
+        raise httpx.ReadError("connection reset")
+
+    async def aclose(self) -> None:
+        pass
+
+
+@pytest.mark.anyio
+async def test_a_non_resumable_sse_drop_resolves_the_request_with_an_error() -> None:
+    """A per-request SSE stream that dies having carried no event ids can never deliver its
+    response; the transport resolves the waiter with CONNECTION_CLOSED instead of hanging forever."""
+    dying = _DyingSSEStream()
+
+    def handler(request: httpx.Request) -> httpx.Response:
+        return httpx.Response(200, headers={"content-type": "text/event-stream"}, stream=dying)
+
+    with anyio.fail_after(5):
+        async with (
+            httpx.AsyncClient(transport=httpx.MockTransport(handler)) as http,
+            streamable_http_client("http://test/mcp", http_client=http) as (read, write),
+        ):
+            await write.send(
+                SessionMessage(JSONRPCRequest(jsonrpc="2.0", id="listen-1", method="subscriptions/listen", params={}))
+            )
+            reply = await read.receive()
+    assert isinstance(reply, SessionMessage)
+    assert isinstance(reply.message, JSONRPCError)
+    assert reply.message.id == "listen-1"
+    assert reply.message.error.code == CONNECTION_CLOSED
+
+
+class _DeliverOnCommandSSEStream(httpx.AsyncByteStream):
+    """Parks after opening, then delivers one JSON-RPC response when told."""
+
+    def __init__(self, response_body: dict[str, Any]) -> None:
+        self._event = f"data: {json.dumps(response_body)}\n\n".encode()
+        self.opened = anyio.Event()
+        self.deliver = anyio.Event()
+
+    async def __aiter__(self) -> AsyncIterator[bytes]:
+        self.opened.set()
+        await self.deliver.wait()
+        yield self._event
+
+    async def aclose(self) -> None:
+        pass
+
+
+@pytest.mark.anyio
+async def test_a_superseded_posts_late_real_response_cannot_answer_the_successor() -> None:
+    """SDK-defined: re-issuing an id severs the superseded POST, so nothing from its
+    stream (a late real response, or a synthesized error for its death) can resolve
+    the reused id's waiter; only the successor's own response arrives."""
+    stale = _DeliverOnCommandSSEStream({"jsonrpc": "2.0", "id": "dup-1", "result": {"origin": "stale"}})
+    succeeding = _DeliverOnCommandSSEStream({"jsonrpc": "2.0", "id": "dup-1", "result": {"origin": "fresh"}})
+    streams: list[httpx.AsyncByteStream] = [stale, succeeding]
+
+    def handler(request: httpx.Request) -> httpx.Response:
+        return httpx.Response(200, headers={"content-type": "text/event-stream"}, stream=streams.pop(0))
+
+    with anyio.fail_after(5):
+        async with (
+            httpx.AsyncClient(transport=httpx.MockTransport(handler)) as http,
+            streamable_http_client("http://test/mcp", http_client=http) as (read, write),
+        ):
+            await write.send(SessionMessage(JSONRPCRequest(jsonrpc="2.0", id="dup-1", method="tools/call", params={})))
+            await stale.opened.wait()
+            await write.send(SessionMessage(JSONRPCRequest(jsonrpc="2.0", id="dup-1", method="tools/call", params={})))
+            await succeeding.opened.wait()
+            stale.deliver.set()
+            await anyio.wait_all_tasks_blocked()
+            succeeding.deliver.set()
+            reply = await read.receive()
+    assert isinstance(reply, SessionMessage)
+    assert isinstance(reply.message, JSONRPCResponse), reply.message
+    assert reply.message.result == {"origin": "fresh"}
+
+
+@pytest.mark.anyio
+async def test_a_202_to_a_request_resolves_the_waiter_with_an_error() -> None:
+    """SDK-defined: a server that answers a request with 202 Accepted has declared no
+    response will follow (the spec requires SSE or JSON for requests); the transport
+    resolves the waiter with INVALID_REQUEST instead of parking the caller forever."""
+
+    def handler(request: httpx.Request) -> httpx.Response:
+        return httpx.Response(202)
+
+    with anyio.fail_after(5):
+        async with (
+            httpx.AsyncClient(transport=httpx.MockTransport(handler)) as http,
+            streamable_http_client("http://test/mcp", http_client=http) as (read, write),
+        ):
+            await write.send(
+                SessionMessage(JSONRPCRequest(jsonrpc="2.0", id="listen-1", method="subscriptions/listen", params={}))
+            )
+            reply = await read.receive()
+    assert isinstance(reply, SessionMessage)
+    assert isinstance(reply.message, JSONRPCError)
+    assert reply.message.id == "listen-1"
+    assert reply.message.error.code == INVALID_REQUEST
+
+
+def _abandoned_request_context(
+    http: httpx.AsyncClient, send: ContextSendStream[SessionMessage | Exception]
+) -> RequestContext:
+    return RequestContext(
+        client=http,
+        session_id=None,
+        session_message=SessionMessage(
+            JSONRPCRequest(jsonrpc="2.0", id="listen-1", method="subscriptions/listen", params={})
+        ),
+        metadata=None,
+        read_stream_writer=send,
+    )
+
+
+@pytest.mark.anyio
+async def test_exhausted_reconnection_attempts_resolve_the_request_with_an_error() -> None:
+    """An id-bearing stream that exhausts its reconnection budget also resolves the waiter with CONNECTION_CLOSED."""
+    transport = StreamableHTTPTransport("http://test/mcp")
+    send, receive = create_context_streams[SessionMessage | Exception](1)
+    async with httpx.AsyncClient() as http:
+        with anyio.fail_after(5):
+            await transport._handle_reconnection(  # pyright: ignore[reportPrivateUsage]
+                _abandoned_request_context(http, send), "evt-7", None, MAX_RECONNECTION_ATTEMPTS
+            )
+            reply = await receive.receive()
+    assert isinstance(reply, SessionMessage)
+    assert isinstance(reply.message, JSONRPCError)
+    assert reply.message.id == "listen-1"
+    assert reply.message.error.code == CONNECTION_CLOSED
+    send.close()
+    receive.close()
+
+
+@pytest.mark.anyio
+async def test_resolving_an_abandoned_request_after_the_reader_closed_is_contained() -> None:
+    """Teardown race: a stream dying after the reader closed resolves best-effort and must not crash."""
+    transport = StreamableHTTPTransport("http://test/mcp")
+    send, receive = create_context_streams[SessionMessage | Exception](1)
+    receive.close()
+    async with httpx.AsyncClient() as http:
+        with anyio.fail_after(5):
+            await transport._handle_reconnection(  # pyright: ignore[reportPrivateUsage]
+                _abandoned_request_context(http, send), "evt-7", None, MAX_RECONNECTION_ATTEMPTS
+            )
+    send.close()

tests/client/test_subscriptions.py

@@ -0,0 +1,666 @@
+"""Behavioral tests for the client-side `subscriptions/listen` driver (SDK-defined contract).
+
+Public API only, against in-process servers; wire-shape assertions live in the interaction suite.
+"""
+
+from itertools import count
+from typing import Any
+
+import anyio
+import mcp_types as types
+import pytest
+from mcp_types import SubscriptionFilter
+
+import mcp.client.subscriptions as subscriptions_module
+from mcp import Client, MCPError
+from mcp.client.session import ClientSession
+from mcp.client.subscriptions import (
+    ListenNotSupportedError,
+    ListenRoute,
+    PromptsListChanged,
+    ResourcesListChanged,
+    ResourceUpdated,
+    ServerEvent,
+    Subscription,
+    SubscriptionLost,
+    ToolsListChanged,
+    listen,
+)
+from mcp.server import Server, ServerRequestContext
+from mcp.server.subscriptions import (
+    SUBSCRIPTION_ID_META_KEY,
+    InMemorySubscriptionBus,
+    ListenHandler,
+)
+from mcp.shared.direct_dispatcher import create_direct_dispatcher_pair
+from mcp.shared.dispatcher import CallOptions
+
+pytestmark = pytest.mark.anyio
+
+
+def _bus_server(bus: InMemorySubscriptionBus, *, max_subscriptions: int | None = None) -> Server[Any]:
+    """A lowlevel server whose only feature is serving listen streams from `bus`."""
+    handler = (
+        ListenHandler(bus) if max_subscriptions is None else ListenHandler(bus, max_subscriptions=max_subscriptions)
+    )
+    return Server("subs", on_subscriptions_listen=handler)
+
+
+async def _ack(ctx: ServerRequestContext[Any, Any], honored: SubscriptionFilter) -> dict[str, Any]:
+    """Send a hand-rolled ack for a scripted listen handler; returns the stamped meta."""
+    assert ctx.request_id is not None
+    meta: dict[str, Any] = {SUBSCRIPTION_ID_META_KEY: ctx.request_id}
+    await ctx.session.send_notification(
+        types.SubscriptionsAcknowledgedNotification(
+            params=types.SubscriptionsAcknowledgedNotificationParams(notifications=honored, _meta=meta)
+        ),
+        related_request_id=ctx.request_id,
+    )
+    return meta
+
+
+async def test_listen_surfaces_the_honored_filter_and_subscription_id():
+    """Entering waits for the server ack and surfaces the honored filter and subscription id."""
+    bus = InMemorySubscriptionBus()
+    async with Client(_bus_server(bus)) as client:
+        with anyio.fail_after(5):
+            async with client.listen(  # pragma: no branch
+                tools_list_changed=True, resource_subscriptions=["note://todo"]
+            ) as sub:
+                assert isinstance(sub, Subscription)
+                assert sub.honored.tools_list_changed is True
+                assert sub.honored.resource_subscriptions == ["note://todo"]
+                assert isinstance(sub.subscription_id, str)
+                assert sub.subscription_id.startswith("listen-")
+
+
+async def test_listen_delivers_all_four_typed_event_kinds():
+    """Bus publishes come back as the same typed event values, in order."""
+    bus = InMemorySubscriptionBus()
+    async with Client(_bus_server(bus)) as client:
+        with anyio.fail_after(5):
+            async with client.listen(  # pragma: no branch
+                tools_list_changed=True,
+                prompts_list_changed=True,
+                resources_list_changed=True,
+                resource_subscriptions=["note://todo"],
+            ) as sub:
+                for event in (
+                    ToolsListChanged(),
+                    PromptsListChanged(),
+                    ResourcesListChanged(),
+                    ResourceUpdated(uri="note://todo"),
+                ):
+                    await bus.publish(event)
+                    assert await anext(sub) == event
+
+
+async def test_unconsumed_duplicate_events_coalesce():
+    """Events are level triggers: duplicates pending consumption collapse to one."""
+    bus = InMemorySubscriptionBus()
+    async with Client(_bus_server(bus)) as client:
+        with anyio.fail_after(5):
+            async with client.listen(  # pragma: no branch
+                tools_list_changed=True, resource_subscriptions=["note://todo"]
+            ) as sub:
+                for _ in range(3):
+                    await bus.publish(ToolsListChanged())
+                await bus.publish(ResourceUpdated(uri="note://todo"))
+                await anyio.wait_all_tasks_blocked()
+                assert await anext(sub) == ToolsListChanged()
+                assert await anext(sub) == ResourceUpdated(uri="note://todo")
+
+
+async def test_graceful_server_close_ends_the_loop_cleanly():
+    """The server's deliberate close ends iteration cleanly, after draining prior events."""
+    bus = InMemorySubscriptionBus()
+    handler = ListenHandler(bus)
+    server = Server("subs", on_subscriptions_listen=handler)
+    events: list[object] = []
+    async with Client(server) as client:
+        with anyio.fail_after(5):
+            async with client.listen(tools_list_changed=True) as sub:  # pragma: no branch
+                await bus.publish(ToolsListChanged())
+                handler.close()
+                events.extend([event async for event in sub])
+    assert events == [ToolsListChanged()]
+
+
+async def test_abrupt_stream_end_raises_subscription_lost():
+    """A stream dying without the graceful result raises `SubscriptionLost` with the cause chained."""
+    proceed = anyio.Event()
+
+    async def dropping_listen(
+        ctx: ServerRequestContext[Any, Any], params: types.SubscriptionsListenRequestParams
+    ) -> types.SubscriptionsListenResult:
+        await _ack(ctx, params.notifications)
+        await proceed.wait()
+        raise MCPError(types.INTERNAL_ERROR, "stream torn down")
+
+    server = Server("subs", on_subscriptions_listen=dropping_listen)
+    async with Client(server) as client:
+        with anyio.fail_after(5):
+            async with client.listen(tools_list_changed=True) as sub:  # pragma: no branch
+                proceed.set()
+                with pytest.raises(SubscriptionLost) as exc_info:  # pragma: no branch
+                    await anext(sub)
+    assert isinstance(exc_info.value.__cause__, MCPError)
+    assert exc_info.value.__cause__.error.message == "stream torn down"
+
+
+async def test_listen_on_a_legacy_connection_raises_the_typed_steer():
+    """On a 2025 connection `listen` fails fast with the typed error steering to the legacy verbs."""
+    bus = InMemorySubscriptionBus()
+    async with Client(_bus_server(bus), mode="legacy") as client:
+        with anyio.fail_after(5):
+            # Entering is where the guard fires; __aenter__ directly avoids an unreachable with-body.
+            with pytest.raises(ListenNotSupportedError) as exc_info:  # pragma: no branch
+                await client.listen(tools_list_changed=True).__aenter__()
+    assert exc_info.value.negotiated_version == "2025-11-25"
+    assert "subscribe_resource" in str(exc_info.value)
+
+
+async def test_server_rejection_raises_from_enter_not_from_iteration():
+    """A server without the listen handler fails the open from entering the context."""
+    server = Server("no-listen")
+    async with Client(server) as client:
+        with anyio.fail_after(5):
+            with pytest.raises(MCPError) as exc_info:  # pragma: no branch
+                await client.listen(tools_list_changed=True).__aenter__()
+    assert exc_info.value.error.code == types.METHOD_NOT_FOUND
+
+
+async def test_immediate_result_without_ack_opens_already_closed():
+    """A bare result with no ack yields a subscription already gracefully over: no filter, no events."""
+
+    async def degenerate_listen(
+        ctx: ServerRequestContext[Any, Any], params: types.SubscriptionsListenRequestParams
+    ) -> types.SubscriptionsListenResult:
+        assert ctx.request_id is not None
+        return types.SubscriptionsListenResult(_meta={SUBSCRIPTION_ID_META_KEY: ctx.request_id})
+
+    server = Server("subs", on_subscriptions_listen=degenerate_listen)
+    async with Client(server) as client:
+        with anyio.fail_after(5):
+            async with client.listen(tools_list_changed=True) as sub:  # pragma: no branch
+                assert sub.honored == SubscriptionFilter()
+                with pytest.raises(StopAsyncIteration):  # pragma: no branch
+                    await anext(sub)
+
+
+async def test_server_sent_cancelled_for_the_listen_id_raises_subscription_lost():
+    """Server-sent notifications/cancelled for the listen id surfaces as a lost subscription."""
+    proceed = anyio.Event()
+
+    async def cancelling_listen(
+        ctx: ServerRequestContext[Any, Any], params: types.SubscriptionsListenRequestParams
+    ) -> types.SubscriptionsListenResult:
+        assert ctx.request_id is not None
+        await _ack(ctx, params.notifications)
+        await proceed.wait()
+        await ctx.session.send_notification(
+            types.CancelledNotification(params=types.CancelledNotificationParams(request_id=ctx.request_id)),
+            related_request_id=ctx.request_id,
+        )
+        await anyio.sleep_forever()
+        raise AssertionError("unreachable")  # pragma: no cover
+
+    server = Server("subs", on_subscriptions_listen=cancelling_listen)
+    async with Client(server) as client:
+        with anyio.fail_after(5):
+            async with client.listen(tools_list_changed=True) as sub:  # pragma: no branch
+                proceed.set()
+                with pytest.raises(SubscriptionLost):  # pragma: no branch
+                    await anext(sub)
+
+
+async def test_exiting_the_context_frees_the_server_slot():
+    """Leaving the block ends the subscription server-side: a one-slot handler admits a second listen."""
+    bus = InMemorySubscriptionBus()
+    async with Client(_bus_server(bus, max_subscriptions=1)) as client:
+        with anyio.fail_after(5):
+            async with client.listen(tools_list_changed=True) as first:
+                assert first.honored.tools_list_changed is True
+            async with client.listen(tools_list_changed=True) as second:  # pragma: no branch
+                assert second.honored.tools_list_changed is True
+                assert second.subscription_id != first.subscription_id
+
+
+async def test_concurrent_subscriptions_demux_independently():
+    """Two open subscriptions each receive only their own filter's events."""
+    bus = InMemorySubscriptionBus()
+    async with Client(_bus_server(bus)) as client:
+        with anyio.fail_after(5):
+            async with (  # pragma: no branch
+                client.listen(tools_list_changed=True) as tools_sub,
+                client.listen(resource_subscriptions=["note://todo"]) as notes_sub,
+            ):
+                await bus.publish(ToolsListChanged())
+                await bus.publish(ResourceUpdated(uri="note://todo"))
+                assert await anext(tools_sub) == ToolsListChanged()
+                assert await anext(notes_sub) == ResourceUpdated(uri="note://todo")
+                # Neither stream received the other's event.
+                await bus.publish(ToolsListChanged())
+                assert await anext(tools_sub) == ToolsListChanged()
+
+
+async def test_change_notifications_still_reach_message_handler():
+    """The demux tees: a delivered event's notification still reaches message_handler; the ack never does."""
+    bus = InMemorySubscriptionBus()
+    seen: list[str] = []
+
+    async def on_message(message: object) -> None:
+        assert not isinstance(message, types.SubscriptionsAcknowledgedNotification)
+        if isinstance(message, types.ToolListChangedNotification):  # pragma: no branch
+            seen.append("tools-changed")
+
+    async with Client(_bus_server(bus), message_handler=on_message) as client:
+        with anyio.fail_after(5):
+            async with client.listen(tools_list_changed=True) as sub:  # pragma: no branch
+                await bus.publish(ToolsListChanged())
+                assert await anext(sub) == ToolsListChanged()
+                await anyio.wait_all_tasks_blocked()
+    assert seen == ["tools-changed"]
+
+
+async def test_enter_times_out_when_the_ack_never_arrives():
+    """The ack wait rides the session's read timeout, so a wedged server cannot hang the open."""
+
+    async def silent_listen(
+        ctx: ServerRequestContext[Any, Any], params: types.SubscriptionsListenRequestParams
+    ) -> types.SubscriptionsListenResult:
+        await anyio.sleep_forever()
+        raise AssertionError("unreachable")  # pragma: no cover
+
+    server = Server("subs", on_subscriptions_listen=silent_listen)
+    async with Client(server, read_timeout_seconds=0.05) as client:
+        with anyio.fail_after(5):
+            with pytest.raises(TimeoutError):  # pragma: no branch
+                await client.listen(tools_list_changed=True).__aenter__()
+
+
+async def test_an_open_stream_outlives_the_session_read_timeout():
+    """The listen request is exempt from the read timeout: the stream delivers after the deadline."""
+    bus = InMemorySubscriptionBus()
+    async with Client(_bus_server(bus), read_timeout_seconds=0.05) as client:
+        with anyio.fail_after(5):
+            async with client.listen(tools_list_changed=True) as sub:  # pragma: no branch
+                # Real clock on purpose: this pins a timeout feature.
+                await anyio.sleep(0.2)
+                await bus.publish(ToolsListChanged())
+                assert await anext(sub) == ToolsListChanged()
+
+
+async def test_a_duplicate_ack_does_not_overwrite_the_honored_filter():
+    """The first ack wins; a later conflicting ack is a no-op."""
+    proceed = anyio.Event()
+
+    async def double_acking_listen(
+        ctx: ServerRequestContext[Any, Any], params: types.SubscriptionsListenRequestParams
+    ) -> types.SubscriptionsListenResult:
+        assert ctx.request_id is not None
+        await _ack(ctx, params.notifications)
+        await _ack(ctx, SubscriptionFilter())
+        await proceed.wait()
+        return types.SubscriptionsListenResult(_meta={SUBSCRIPTION_ID_META_KEY: ctx.request_id})
+
+    server = Server("subs", on_subscriptions_listen=double_acking_listen)
+    async with Client(server) as client:
+        with anyio.fail_after(5):
+            async with client.listen(tools_list_changed=True) as sub:  # pragma: no branch
+                assert sub.honored.tools_list_changed is True
+                proceed.set()
+
+
+async def test_a_non_event_frame_with_the_subscription_id_is_teed_not_delivered():
+    """A stamped non-event notification never surfaces as an event; it flows to message_handler."""
+    proceed = anyio.Event()
+
+    async def logging_listen(
+        ctx: ServerRequestContext[Any, Any], params: types.SubscriptionsListenRequestParams
+    ) -> types.SubscriptionsListenResult:
+        assert ctx.request_id is not None
+        meta = await _ack(ctx, params.notifications)
+        await ctx.session.send_notification(
+            types.LoggingMessageNotification(
+                params=types.LoggingMessageNotificationParams(level="info", data="not an event", _meta=meta)
+            ),
+            related_request_id=ctx.request_id,
+        )
+        await proceed.wait()
+        return types.SubscriptionsListenResult(_meta=meta)
+
+    logged: list[str] = []
+
+    async def on_message(message: object) -> None:
+        if isinstance(message, types.LoggingMessageNotification):  # pragma: no branch
+            logged.append(str(message.params.data))
+
+    server = Server("subs", on_subscriptions_listen=logging_listen)
+    async with Client(server, message_handler=on_message) as client:
+        with anyio.fail_after(5):
+            async with client.listen(tools_list_changed=True) as sub:  # pragma: no branch
+                await anyio.wait_all_tasks_blocked()
+                proceed.set()
+                with pytest.raises(StopAsyncIteration):  # pragma: no branch
+                    await anext(sub)
+    assert logged == ["not an event"]
+
+
+async def test_session_teardown_unblocks_a_sibling_consumer_with_subscription_lost():
+    """Session teardown settles every open route as lost, unblocking parked consumers."""
+    bus = InMemorySubscriptionBus()
+    outcome: list[str] = []
+    entered = anyio.Event()
+
+    async def consume(client: Client) -> None:
+        with pytest.raises(SubscriptionLost):
+            async with client.listen(tools_list_changed=True) as sub:
+                entered.set()
+                await anext(sub)
+        outcome.append("lost")
+
+    with anyio.fail_after(5):
+        async with anyio.create_task_group() as tg:
+            async with Client(_bus_server(bus)) as client:  # pragma: no branch
+                tg.start_soon(consume, client)
+                await entered.wait()
+    assert outcome == ["lost"]
+
+
+async def test_server_cancel_before_the_ack_raises_subscription_lost_from_enter():
+    """A stream torn down before it was ever acknowledged is a failed open: enter raises."""
+
+    async def cancel_first_listen(
+        ctx: ServerRequestContext[Any, Any], params: types.SubscriptionsListenRequestParams
+    ) -> types.SubscriptionsListenResult:
+        assert ctx.request_id is not None
+        await ctx.session.send_notification(
+            types.CancelledNotification(params=types.CancelledNotificationParams(request_id=ctx.request_id)),
+            related_request_id=ctx.request_id,
+        )
+        await anyio.sleep_forever()
+        raise AssertionError("unreachable")  # pragma: no cover
+
+    server = Server("subs", on_subscriptions_listen=cancel_first_listen)
+    async with Client(server) as client:
+        with anyio.fail_after(5):
+            with pytest.raises(SubscriptionLost, match="before it was acknowledged"):  # pragma: no branch
+                await client.listen(tools_list_changed=True).__aenter__()
+
+
+async def test_listen_on_an_exited_session_raises_and_leaks_no_route():
+    """Opening on an exited session fails loudly and leaves no demux registration behind."""
+    bus = InMemorySubscriptionBus()
+    client = Client(_bus_server(bus))
+    async with client:
+        session = client.session
+    with pytest.raises(RuntimeError):
+        await listen(session, tools_list_changed=True).__aenter__()
+    assert session._listen_routes == {}  # pyright: ignore[reportPrivateUsage]
+
+
+async def test_listen_on_a_never_entered_session_raises_runtime_error():
+    """An adopted-but-never-entered session has no task group to drive the stream."""
+    dispatcher, _peer = create_direct_dispatcher_pair()
+    session = ClientSession(dispatcher=dispatcher)
+    session.adopt(
+        types.DiscoverResult(
+            supported_versions=["2026-07-28"],
+            capabilities=types.ServerCapabilities(),
+            server_info=types.Implementation(name="stub", version="0"),
+        )
+    )
+    with pytest.raises(RuntimeError, match="entered session"):
+        await listen(session, tools_list_changed=True).__aenter__()
+    assert session._listen_routes == {}  # pyright: ignore[reportPrivateUsage]
+
+
+async def test_a_retained_handle_after_exit_does_not_serve_stale_events():
+    """Leaving the block abandons the backlog: a stashed handle must not replay buffered events."""
+    bus = InMemorySubscriptionBus()
+    async with Client(_bus_server(bus)) as client:
+        with anyio.fail_after(5):
+            async with client.listen(tools_list_changed=True) as sub:
+                await bus.publish(ToolsListChanged())
+                await anyio.wait_all_tasks_blocked()
+            with pytest.raises(StopAsyncIteration):  # pragma: no branch
+                await anext(sub)
+
+
+async def test_a_stray_ack_outside_the_driver_namespace_still_reaches_message_handler():
+    """Acks for ids the driver never minted flow to message_handler (the raw-listen escape hatch)."""
+    proceed = anyio.Event()
+
+    async def stray_acking_listen(
+        ctx: ServerRequestContext[Any, Any], params: types.SubscriptionsListenRequestParams
+    ) -> types.SubscriptionsListenResult:
+        assert ctx.request_id is not None
+        await _ack(ctx, params.notifications)
+        await ctx.session.send_notification(
+            types.SubscriptionsAcknowledgedNotification(
+                params=types.SubscriptionsAcknowledgedNotificationParams(
+                    notifications=SubscriptionFilter(), _meta={SUBSCRIPTION_ID_META_KEY: 424242}
+                )
+            ),
+            related_request_id=ctx.request_id,
+        )
+        await proceed.wait()
+        return types.SubscriptionsListenResult(_meta={SUBSCRIPTION_ID_META_KEY: ctx.request_id})
+
+    handled: list[str] = []
+
+    async def on_message(message: object) -> None:
+        handled.append(type(message).__name__)
+
+    server = Server("subs", on_subscriptions_listen=stray_acking_listen)
+    async with Client(server, message_handler=on_message) as client:
+        with anyio.fail_after(5):
+            async with client.listen(tools_list_changed=True) as sub:  # pragma: no branch
+                await anyio.wait_all_tasks_blocked()
+                proceed.set()
+                with pytest.raises(StopAsyncIteration):  # pragma: no branch
+                    await anext(sub)
+    assert "SubscriptionsAcknowledgedNotification" in handled
+
+
+async def test_a_bare_string_for_resource_subscriptions_is_rejected():
+    """A bare string would explode into per-character URIs; it is rejected before touching the wire."""
+    bus = InMemorySubscriptionBus()
+    async with Client(_bus_server(bus)) as client:
+        with pytest.raises(TypeError, match="sequence of URIs"):
+            await client.listen(resource_subscriptions="note://todo").__aenter__()  # pyright: ignore[reportArgumentType]
+
+
+def test_the_route_admits_only_honored_events_and_only_while_live():
+    """Route admission: nothing before the ack, only honored events while live, nothing after the end."""
+    route = ListenRoute()
+    route.deliver(ToolsListChanged())
+    assert route._pending == {}  # pyright: ignore[reportPrivateUsage]
+    route.set_acked(SubscriptionFilter(tools_list_changed=True, resource_subscriptions=["note://todo"]))
+    route.deliver(PromptsListChanged())  # kind not honored
+    route.deliver(ResourceUpdated(uri="note://todo/draft"))  # sub-resource of a subscribed URI: spec says admit
+    route.deliver(ResourceUpdated(uri="note://todo"))
+    route.deliver(ToolsListChanged())
+    route.deliver(ToolsListChanged())  # duplicate pending consumption collapses
+    assert list(route._pending) == [  # pyright: ignore[reportPrivateUsage]
+        ResourceUpdated(uri="note://todo/draft"),
+        ResourceUpdated(uri="note://todo"),
+        ToolsListChanged(),
+    ]
+    route.settle("graceful")
+    route.deliver(ResourceUpdated(uri="note://todo"))  # post-close noise is refused
+    assert len(route._pending) == 3  # pyright: ignore[reportPrivateUsage]
+
+
+def test_a_peer_flooding_distinct_uris_costs_the_subscription_not_client_memory():
+    """A peer flooding distinct URIs trips the `_MAX_PENDING_EVENTS` backstop: the route
+    settles lost instead of growing client memory without bound."""
+    route = ListenRoute()
+    route.set_acked(SubscriptionFilter(resource_subscriptions=["note://todo"]))
+    for n in range(subscriptions_module._MAX_PENDING_EVENTS):  # pyright: ignore[reportPrivateUsage]
+        route.deliver(ResourceUpdated(uri=f"note://todo/{n}"))
+    assert route.end is None
+    route.deliver(ResourceUpdated(uri="note://todo/one-too-many"))
+    assert route.end == "lost"
+    assert route.error is not None
+    assert "backlog" in route.error.error.message
+    # The overflowing event was not queued.
+    assert len(route._pending) == subscriptions_module._MAX_PENDING_EVENTS  # pyright: ignore[reportPrivateUsage]
+
+
+async def test_a_cancelled_on_event_barrier_does_not_lose_the_event():
+    """Cancelling `anext` mid-barrier leaves the event queued; the next `anext` re-runs the
+    idempotent barrier and returns it."""
+    bus = InMemorySubscriptionBus()
+    entered = anyio.Event()
+    release = anyio.Event()
+
+    async def parked_barrier(event: ServerEvent) -> None:
+        entered.set()
+        await release.wait()
+
+    async with Client(_bus_server(bus)) as client:
+        with anyio.fail_after(5):
+            async with listen(
+                client.session, tools_list_changed=True, on_event=parked_barrier
+            ) as sub:  # pragma: no branch
+                await bus.publish(ToolsListChanged())
+                async with anyio.create_task_group() as tg:
+                    cancel_scope = anyio.CancelScope()
+
+                    async def first_attempt() -> None:
+                        with cancel_scope:
+                            await anext(sub)
+                            raise AssertionError("must be cancelled mid-barrier")  # pragma: no cover
+
+                    tg.start_soon(first_attempt)
+                    await entered.wait()
+                    cancel_scope.cancel()
+                release.set()
+                assert await anext(sub) == ToolsListChanged()
+
+
+async def test_events_outside_the_honored_filter_are_never_delivered():
+    """A server violating its acknowledged filter cannot reach the consumer or grow the backlog."""
+    proceed = anyio.Event()
+
+    async def overreaching_listen(
+        ctx: ServerRequestContext[Any, Any], params: types.SubscriptionsListenRequestParams
+    ) -> types.SubscriptionsListenResult:
+        meta = await _ack(ctx, params.notifications)  # honors exactly what was requested: tools only
+        await ctx.session.send_notification(
+            types.ResourceUpdatedNotification(
+                params=types.ResourceUpdatedNotificationParams(uri="note://uninvited", _meta=meta)
+            ),
+            related_request_id=ctx.request_id,
+        )
+        await ctx.session.send_notification(
+            types.ToolListChangedNotification(params=types.NotificationParams(_meta=meta)),
+            related_request_id=ctx.request_id,
+        )
+        await proceed.wait()
+        return types.SubscriptionsListenResult(_meta=meta)
+
+    server = Server("subs", on_subscriptions_listen=overreaching_listen)
+    async with Client(server) as client:
+        with anyio.fail_after(5):
+            async with client.listen(tools_list_changed=True) as sub:  # pragma: no branch
+                assert await anext(sub) == ToolsListChanged()
+                proceed.set()
+                with pytest.raises(StopAsyncIteration):  # pragma: no branch
+                    await anext(sub)
+
+
+async def test_the_on_event_barrier_completes_before_each_event_is_returned():
+    """`on_event` is awaited before the iterator returns each event (the Client wires cache eviction here)."""
+    bus = InMemorySubscriptionBus()
+    order: list[str] = []
+
+    async def barrier(event: ServerEvent) -> None:
+        order.append(f"barrier:{type(event).__name__}")
+
+    async with Client(_bus_server(bus)) as client:
+        with anyio.fail_after(5):
+            async with listen(client.session, tools_list_changed=True, on_event=barrier) as sub:  # pragma: no branch
+                await bus.publish(ToolsListChanged())
+                event = await anext(sub)
+                order.append(f"returned:{type(event).__name__}")
+    assert order == ["barrier:ToolsListChanged", "returned:ToolsListChanged"]
+
+
+async def test_client_listen_installs_the_cache_eviction_barrier_exactly_when_a_cache_exists():
+    """`Client.listen` wires the response-cache evictor as the barrier only when a cache exists."""
+    bus = InMemorySubscriptionBus()
+    async with Client(_bus_server(bus)) as cached_client:
+        with anyio.fail_after(5):
+            async with cached_client.listen(tools_list_changed=True) as sub:  # pragma: no branch
+                assert sub._on_event == cached_client._evict_for_listen_event  # pyright: ignore[reportPrivateUsage]
+    async with Client(_bus_server(bus), cache=False) as uncached_client:
+        with anyio.fail_after(5):
+            async with uncached_client.listen(tools_list_changed=True) as sub:  # pragma: no branch
+                assert sub._on_event is None  # pyright: ignore[reportPrivateUsage]
+
+
+async def test_the_cache_eviction_barrier_maps_events_and_contains_store_faults(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """The barrier evicts through the same notification mapping as the message_handler wrapper;
+    a raising store costs a log line, not the delivery."""
+    client = Client(_bus_server(InMemorySubscriptionBus()))
+    cache = client._response_cache  # pyright: ignore[reportPrivateUsage]
+    assert cache is not None
+    evicted: list[types.ServerNotification] = []
+
+    async def record(notification: types.ServerNotification) -> None:
+        evicted.append(notification)
+
+    monkeypatch.setattr(cache, "evict_for_notification", record)
+    await client._evict_for_listen_event(ResourceUpdated(uri="note://x"))  # pyright: ignore[reportPrivateUsage]
+    assert isinstance(evicted[0], types.ResourceUpdatedNotification)
+    assert evicted[0].params.uri == "note://x"
+
+    async def broken(notification: types.ServerNotification) -> None:
+        raise RuntimeError("store down")
+
+    monkeypatch.setattr(cache, "evict_for_notification", broken)
+    # Contained: a cache fault must not block delivery.
+    await client._evict_for_listen_event(ToolsListChanged())  # pyright: ignore[reportPrivateUsage]
+
+
+async def test_a_raw_request_id_collision_fails_the_subscription_not_the_session(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """A raw caller occupying the driver's next minted id fails that one listen from enter;
+    the session survives and the next listen opens normally."""
+    monkeypatch.setattr(subscriptions_module, "_listen_ids", count(7000))
+    bus = InMemorySubscriptionBus()
+    async with Client(_bus_server(bus)) as client:
+        with anyio.fail_after(5):
+            async with anyio.create_task_group() as tg:  # pragma: no branch
+                raw_scope = anyio.CancelScope()
+
+                async def raw_listen() -> None:
+                    request = types.SubscriptionsListenRequest(
+                        params=types.SubscriptionsListenRequestParams(
+                            notifications=SubscriptionFilter(tools_list_changed=True)
+                        )
+                    )
+                    data = request.model_dump(by_alias=True, mode="json", exclude_none=True)
+                    opts: CallOptions = {"request_id": "listen-7000"}
+                    client.session._stamp(data, opts)  # pyright: ignore[reportPrivateUsage]
+                    with raw_scope:
+                        await client.session._dispatcher.send_raw_request(  # pyright: ignore[reportPrivateUsage]
+                            data["method"], data.get("params"), opts
+                        )
+
+                tg.start_soon(raw_listen)
+                await anyio.wait_all_tasks_blocked()
+                with pytest.raises(MCPError) as exc_info:
+                    await client.listen(tools_list_changed=True).__aenter__()
+                assert "already in flight" in exc_info.value.error.message
+                # The failed open released the colliding id's demux registration.
+                assert client.session._listen_routes == {}  # pyright: ignore[reportPrivateUsage]
+                raw_scope.cancel()
+                async with client.listen(tools_list_changed=True) as sub:  # pragma: no branch
+                    assert sub.subscription_id == "listen-7001"

tests/docs_src/test_client.py

@@ -5,7 +5,7 @@ from inline_snapshot import snapshot
 from mcp_types import Prompt, PromptArgument, PromptReference, TextContent, TextResourceContents, Tool
 
 from docs_src.client import tutorial001, tutorial002, tutorial003, tutorial004, tutorial005, tutorial006, tutorial007
-from mcp import Client, MCPError
+from mcp import Client, MCPDeprecationWarning, MCPError
 from mcp.shared.metadata_utils import get_display_name
 
 # See test_index.py for why this is a per-module mark and not a conftest hook.
@@ -128,7 +128,9 @@ async def test_resource_subscriptions_are_listen_based_on_the_modern_wire() -> N
         assert client.server_capabilities.resources is not None
         assert client.server_capabilities.resources.subscribe is True
         with pytest.raises(MCPError) as exc_info:
-            await client.subscribe_resource("catalog://genres")
+            # The verb is itself deprecated; the modern wire also rejects it.
+            with pytest.warns(MCPDeprecationWarning, match="use Client.listen"):
+                await client.subscribe_resource("catalog://genres")  # pyright: ignore[reportDeprecated]
         assert exc_info.value.error.code == -32601
         assert exc_info.value.error.message == "Method not found"
 

tests/docs_src/test_subscriptions.py

@@ -1,19 +1,40 @@
-"""`docs/handlers/subscriptions.md`: every claim the page makes, proved against the real SDK."""
+"""`docs/{handlers,client}/subscriptions.md`: every claim the two pages make, proved against the real SDK."""
 
+from collections.abc import Awaitable, Callable
 from typing import Any
 
 import anyio
 import mcp_types as types
 import pytest
+from trio.testing import MockClock
 
-from docs_src.subscriptions import tutorial001, tutorial002
+from docs_src.subscriptions import (
+    tutorial001,
+    tutorial002,
+    tutorial003,
+    tutorial004_anyio,
+    tutorial004_asyncio,
+    tutorial004_trio,
+    tutorial005,
+)
 from mcp import Client
-from mcp.server.subscriptions import SUBSCRIPTION_ID_META_KEY, ToolsListChanged
+from mcp.server.context import ServerRequestContext
+from mcp.server.lowlevel import Server
+from mcp.server.subscriptions import SUBSCRIPTION_ID_META_KEY, ListenHandler, ToolsListChanged
+
+_ReadResource = Callable[
+    [ServerRequestContext[Any], types.ReadResourceRequestParams], Awaitable[types.ReadResourceResult]
+]
 
 # See test_index.py for why this is a per-module mark and not a conftest hook.
 pytestmark = [pytest.mark.anyio, pytest.mark.filterwarnings("error::mcp.MCPDeprecationWarning")]
 
 
+@pytest.fixture(autouse=True)
+def _module_runner_lease() -> None:
+    """Opt out of the shared per-module event loop: this module parametrizes `anyio_backend`."""
+
+
 class _Stream:
     """Collects listen-stream notifications and lets tests await arrival counts."""
 
@@ -42,12 +63,56 @@ class _Stream:
                 await self._arrival.wait()
 
 
+class _Reads:
+    """Counts server-side resource reads so a test can await the Nth refetch."""
+
+    def __init__(self) -> None:
+        self.count = 0
+        self._bump = anyio.Event()
+
+    def counting(self, handler: _ReadResource) -> _ReadResource:
+        async def counted(
+            ctx: ServerRequestContext[Any], params: types.ReadResourceRequestParams
+        ) -> types.ReadResourceResult:
+            result = await handler(ctx, params)
+            self.count += 1
+            self._bump.set()
+            self._bump = anyio.Event()
+            return result
+
+        return counted
+
+    async def wait_for(self, count: int) -> None:
+        with anyio.fail_after(5):
+            while self.count < count:
+                await self._bump.wait()
+
+
 def _listen_request(**fields: Any) -> types.SubscriptionsListenRequest:
     return types.SubscriptionsListenRequest(
         params=types.SubscriptionsListenRequestParams(notifications=types.SubscriptionFilter(**fields))
     )
 
 
+@pytest.fixture(autouse=True)
+def _fresh_server_state() -> Any:
+    """Each test starts from an all-unfinished board and the base tool set.
+
+    The tutorials mutate module state deliberately (that is what publishes events), so the
+    board contents and the `enable_reports` registration have to be undone between tests.
+    """
+    boards = {name: dict(tasks) for name, tasks in tutorial001.BOARDS.items()}
+    lowlevel_board = dict(tutorial002.BOARD)
+    tools = dict(tutorial001.mcp._tool_manager._tools)  # pyright: ignore[reportPrivateUsage]
+    yield
+    tutorial001.BOARDS.clear()
+    tutorial001.BOARDS.update(boards)
+    tutorial002.BOARD.clear()
+    tutorial002.BOARD.update(lowlevel_board)
+    tutorial001.mcp._tool_manager._tools.clear()  # pyright: ignore[reportPrivateUsage]
+    tutorial001.mcp._tool_manager._tools.update(tools)  # pyright: ignore[reportPrivateUsage]
+
+
 async def test_publishes_reach_the_stream_filtered_and_tagged() -> None:
     """tutorial001: the full arc - ack first, exact-URI filtering, list_changed
     leading to a refreshed tool list, and client-side close."""
@@ -57,7 +122,7 @@ async def test_publishes_reach_the_stream_filtered_and_tagged() -> None:
 
             async def listen() -> None:
                 await client.session.send_request(
-                    _listen_request(tools_list_changed=True, resource_subscriptions=["note://todo"]),
+                    _listen_request(tools_list_changed=True, resource_subscriptions=["board://sprint"]),
                     types.SubscriptionsListenResult,
                 )
 
@@ -67,21 +132,21 @@ async def test_publishes_reach_the_stream_filtered_and_tagged() -> None:
             ack = stream.received[0]
             assert isinstance(ack, types.SubscriptionsAcknowledgedNotification)
             assert ack.params.notifications == types.SubscriptionFilter(
-                tools_list_changed=True, resource_subscriptions=["note://todo"]
+                tools_list_changed=True, resource_subscriptions=["board://sprint"]
             )
             assert ack.params.meta is not None and SUBSCRIPTION_ID_META_KEY in ack.params.meta
 
             # An edit to a URI the stream did not subscribe to stays silent...
-            await client.call_tool("edit_note", {"name": "journal", "text": "day two"})
+            await client.call_tool("complete_task", {"board": "backlog", "task": "tidy docs"})
             # ...and the subscribed URI delivers, tagged with the same subscription id.
-            await client.call_tool("edit_note", {"name": "todo", "text": "water plants"})
+            await client.call_tool("complete_task", {"board": "sprint", "task": "design"})
             await stream.wait_for(2)
             updated = stream.received[1]
             assert isinstance(updated, types.ResourceUpdatedNotification)
-            assert updated.params.uri == "note://todo"
+            assert updated.params.uri == "board://sprint"
             assert updated.params.meta == ack.params.meta
 
-            await client.call_tool("enable_search", {})
+            await client.call_tool("enable_reports", {})
             await stream.wait_for(3)
             assert isinstance(stream.received[2], types.ToolListChangedNotification)
 
@@ -91,16 +156,16 @@ async def test_publishes_reach_the_stream_filtered_and_tagged() -> None:
         # The list_changed told us to re-fetch: the new tool is there, and the
         # session outlives the closed stream.
         tools = await client.list_tools()
-        assert "search" in {tool.name for tool in tools.tools}
-        contents = (await client.read_resource("note://todo")).contents[0]
+        assert "sprint_report" in {tool.name for tool in tools.tools}
+        contents = (await client.read_resource("board://sprint")).contents[0]
         assert isinstance(contents, types.TextResourceContents)
-        assert contents.text == "water plants"
+        assert contents.text == "[x] design\n[ ] build\n[ ] ship"
 
 
 async def test_publish_with_no_subscribers_is_a_no_op() -> None:
     """tutorial001: publishing to an idle server does nothing and breaks nothing."""
     async with Client(tutorial001.mcp, mode="2026-07-28") as client:
-        result = await client.call_tool("edit_note", {"name": "todo", "text": "buy milk"})
+        result = await client.call_tool("complete_task", {"board": "sprint", "task": "design"})
         assert result.is_error is not True
 
 
@@ -109,30 +174,130 @@ async def test_lowlevel_composition_serves_the_same_stream() -> None:
     stream = _Stream()
     async with Client(tutorial002.server, mode="2026-07-28", message_handler=stream.handler) as client:
         tools = await client.list_tools()
-        assert [tool.name for tool in tools.tools] == ["edit_note"]
+        assert [tool.name for tool in tools.tools] == ["complete_task"]
 
         async with anyio.create_task_group() as tg:
 
             async def listen() -> None:
                 await client.session.send_request(
-                    _listen_request(resource_subscriptions=["note://todo"]),
+                    _listen_request(resource_subscriptions=["board://sprint"]),
                     types.SubscriptionsListenResult,
                 )
 
             tg.start_soon(listen)
             await stream.wait_for(1)
 
-            await client.call_tool("edit_note", {"name": "todo", "text": "water plants"})
+            await client.call_tool("complete_task", {"task": "design"})
             await stream.wait_for(2)
             updated = stream.received[1]
             assert isinstance(updated, types.ResourceUpdatedNotification)
-            assert updated.params.uri == "note://todo"
+            assert updated.params.uri == "board://sprint"
 
             # The bus you constructed is also the publish surface outside a
             # request; an unrequested kind never reaches this stream.
             await tutorial002.bus.publish(ToolsListChanged())
-            await client.call_tool("edit_note", {"name": "todo", "text": "done"})
+            await client.call_tool("complete_task", {"task": "build"})
             await stream.wait_for(3)
             assert isinstance(stream.received[2], types.ResourceUpdatedNotification)
 
             tg.cancel_scope.cancel()
+
+
+async def test_follow_board_prints_the_refetched_board_and_the_new_tool_list(
+    capsys: pytest.CaptureFixture[str],
+) -> None:
+    """tutorial003: each event drives a refetch - the board reprints, and a tools change reprints the tool names."""
+    async with Client(tutorial001.mcp) as client:
+        async with anyio.create_task_group() as tg:
+            tg.start_soon(tutorial003.follow_board, client)
+            # Let the watcher park on its stream (ack complete) before publishing.
+            await anyio.wait_all_tasks_blocked()
+            await client.call_tool("complete_task", {"board": "sprint", "task": "design"})
+            await anyio.wait_all_tasks_blocked()
+            await client.call_tool("enable_reports", {})
+            await anyio.wait_all_tasks_blocked()
+            tg.cancel_scope.cancel()
+
+    printed = capsys.readouterr().out
+    assert "[x] design\n[ ] build\n[ ] ship" in printed
+    assert "sprint_report" in printed
+
+
+EMPTY_BOARD = "[ ] design\n[ ] build\n[ ] ship"
+FINISHED_BOARD = "[x] design\n[x] build\n[x] ship"
+
+
+def _assert_snapshot_then_current_board(printed: str) -> None:
+    """The snapshot taken inside the open subscription came first, and the watcher ended up current.
+
+    How many times the watcher printed is deliberately not asserted: identical events that pile up
+    unconsumed coalesce, so a fast main flow can turn three completions into one refetch. What the
+    stream guarantees is that no change after the acknowledgment is missed.
+    """
+    assert printed.startswith(EMPTY_BOARD), printed
+    assert printed.strip().endswith(FINISHED_BOARD), printed
+
+
+async def test_the_asyncio_watcher_runs_beside_the_main_flow(capsys: pytest.CaptureFixture[str]) -> None:
+    """tutorial004 (asyncio tab): run_sprint opens the subscription, snapshots the board, then a watcher
+    task reprints it while the main flow keeps calling tools.
+
+    The example connects over HTTP; the in-memory client here is the maintainer-side stand-in."""
+    async with Client(tutorial001.mcp) as client:
+        await tutorial004_asyncio.run_sprint(client)
+    _assert_snapshot_then_current_board(capsys.readouterr().out)
+
+
+@pytest.mark.parametrize("anyio_backend", [pytest.param("trio", id="trio")])
+async def test_the_trio_watcher_runs_beside_the_main_flow(capsys: pytest.CaptureFixture[str]) -> None:
+    """tutorial004 (trio tab): the same shape as the asyncio tab, with a nursery owning the watcher."""
+    async with Client(tutorial001.mcp) as client:
+        await tutorial004_trio.run_sprint(client)
+    _assert_snapshot_then_current_board(capsys.readouterr().out)
+
+
+async def test_the_anyio_watcher_runs_beside_the_main_flow(capsys: pytest.CaptureFixture[str]) -> None:
+    """tutorial004 (anyio tab): the same shape again, with a task group owning the watcher."""
+    async with Client(tutorial001.mcp) as client:
+        await tutorial004_anyio.run_sprint(client)
+    _assert_snapshot_then_current_board(capsys.readouterr().out)
+
+
+@pytest.mark.parametrize(
+    "anyio_backend",
+    [pytest.param(("trio", {"clock": MockClock(autojump_threshold=0)}), id="trio-mockclock")],
+)
+async def test_the_follower_re_listens_after_the_stream_ends(capsys: pytest.CaptureFixture[str]) -> None:
+    """tutorial005: a graceful server close ends one stream; the loop backs off, re-listens, and refetches.
+
+    Runs on trio's autojumping MockClock so the loop's backoff sleep takes no wall-clock time.
+    """
+    reads = _Reads()
+    handler = ListenHandler(tutorial002.bus)
+    server = Server(
+        "sprint-board",
+        on_read_resource=reads.counting(tutorial002.read_resource),
+        on_list_tools=tutorial002.list_tools,
+        on_call_tool=tutorial002.call_tool,
+        on_subscriptions_listen=handler,
+    )
+
+    async with Client(server) as client:
+        async with anyio.create_task_group() as tg:
+            tg.start_soon(tutorial005.keep_following, client)
+            # First stream: the entry refetch reads the board, then an event reads it again.
+            await reads.wait_for(1)
+            await client.call_tool("complete_task", {"task": "design"})
+            await reads.wait_for(2)
+
+            # End that stream gracefully. The loop backs off (the mock clock jumps the
+            # sleep), re-listens, and refetches on entry: that is the third read.
+            handler.close()
+            await reads.wait_for(3)
+            await client.call_tool("complete_task", {"task": "build"})
+            await reads.wait_for(4)
+            tg.cancel_scope.cancel()
+
+    printed = capsys.readouterr().out
+    assert "[x] design\n[ ] build" in printed  # first stream, after design
+    assert "[x] design\n[x] build" in printed  # second stream, after build

tests/interaction/_requirements.py

@@ -486,12 +486,6 @@ REQUIREMENTS: dict[str, Requirement] = {
             "id up front is the SDK surface that makes it satisfiable."
         ),
         added_in="2026-07-28",
-        deferred=(
-            "No public API surface yet: the capability exists at the dispatcher seam "
-            "(CallOptions['request_id'], unit-tested there), but ClientSession.send_request does not "
-            "expose it. The public consumer arrives with the client-side listen driver (Client.listen), "
-            "whose interaction tests will exercise it end to end."
-        ),
     ),
     "protocol:notifications:no-response": Requirement(
         source=f"{SPEC_BASE_URL}/basic#notifications",
@@ -1229,6 +1223,59 @@ REQUIREMENTS: dict[str, Requirement] = {
         removed_in="2026-07-28",
         note="removed in 2026-07-28 (SEP-2575); resources/unsubscribe replaced by subscriptions/listen.",
     ),
+    "subscriptions:listen:client:honored-surfacing": Requirement(
+        source=f"{SPEC_2026_BASE_URL}/basic/patterns/subscriptions#acknowledgment",
+        behavior=(
+            "Entering Client.listen() waits for the server's acknowledgment and surfaces the honored "
+            "filter subset on the handle, so the client can check it against what it requested (spec SHOULD)."
+        ),
+        added_in="2026-07-28",
+    ),
+    "subscriptions:listen:client:concurrent-demux": Requirement(
+        source=f"{SPEC_2026_BASE_URL}/basic/patterns/subscriptions#multiple-concurrent-subscriptions",
+        behavior=(
+            "Concurrently open subscriptions each surface their own acknowledgment: with both listen "
+            "requests in flight before either ack arrives, each handle's honored filter is the subset "
+            "for its own request, routed by subscription id rather than broadcast to every open route."
+        ),
+        added_in="2026-07-28",
+    ),
+    "subscriptions:listen:client:iteration": Requirement(
+        source="sdk",
+        behavior=(
+            "An open subscription is an async iterator of typed change events; delivered notifications "
+            "still tee to message_handler so caching and observers keep working."
+        ),
+        added_in="2026-07-28",
+    ),
+    "subscriptions:listen:client:graceful-close": Requirement(
+        source=f"{SPEC_2026_BASE_URL}/basic/patterns/subscriptions#cancellation",
+        behavior=(
+            "The server's empty subscriptions/listen result (its deliberate close) ends iteration cleanly "
+            "after buffered events drain; no exception is raised."
+        ),
+        added_in="2026-07-28",
+    ),
+    "subscriptions:listen:client:lost": Requirement(
+        source="sdk",
+        behavior=(
+            "A listen stream that ends without the graceful result raises SubscriptionLost from iteration; "
+            "there is no automatic re-listen."
+        ),
+        added_in="2026-07-28",
+    ),
+    "subscriptions:listen:client:era-guard": Requirement(
+        source="sdk",
+        behavior=(
+            "Client.listen() on a pre-2026 connection raises ListenNotSupportedError steering to "
+            "subscribe_resource/message_handler instead of leaking a wire -32601."
+        ),
+        removed_in="2026-07-28",
+        note=(
+            "removed_in scopes the matrix to the 2025 cells deliberately: the behavior under test is the "
+            "guard on connections where the method does not exist."
+        ),
+    ),
     "resources:updated-notification": Requirement(
         source=f"{SPEC_BASE_URL}/server/resources#subscriptions",
         behavior=(

tests/interaction/lowlevel/test_resources.py

@@ -203,6 +203,7 @@ async def test_list_resource_templates_returns_registered_templates(connect: Con
     )
 
 
+@pytest.mark.filterwarnings("ignore::mcp.MCPDeprecationWarning")
 @requirement("resources:subscribe")
 async def test_subscribe_resource_delivers_uri_to_handler(connect: Connect) -> None:
     """Subscribing to a resource delivers the URI to the server's subscribe handler and returns an empty result."""
@@ -214,11 +215,12 @@ async def test_subscribe_resource_delivers_uri_to_handler(connect: Connect) -> N
     server = Server("library", on_subscribe_resource=subscribe_resource)
 
     async with connect(server) as client:
-        result = await client.subscribe_resource("file:///watched.txt")
+        result = await client.subscribe_resource("file:///watched.txt")  # pyright: ignore[reportDeprecated]
 
     assert result == snapshot(EmptyResult())
 
 
+@pytest.mark.filterwarnings("ignore::mcp.MCPDeprecationWarning")
 @requirement("resources:subscribe:capability-required")
 async def test_subscribe_without_a_subscribe_handler_is_method_not_found(connect: Connect) -> None:
     """Subscribing to a server that registered no subscribe handler is rejected with METHOD_NOT_FOUND.
@@ -237,13 +239,14 @@ async def test_subscribe_without_a_subscribe_handler_is_method_not_found(connect
 
     async with connect(server) as client:
         with pytest.raises(MCPError) as exc_info:
-            await client.subscribe_resource("file:///watched.txt")
+            await client.subscribe_resource("file:///watched.txt")  # pyright: ignore[reportDeprecated]
 
     assert exc_info.value.error == snapshot(
         ErrorData(code=METHOD_NOT_FOUND, message="Method not found", data="resources/subscribe")
     )
 
 
+@pytest.mark.filterwarnings("ignore::mcp.MCPDeprecationWarning")
 @requirement("resources:unsubscribe")
 async def test_unsubscribe_resource_delivers_uri_to_handler(connect: Connect) -> None:
     """Unsubscribing from a resource delivers the URI to the server's unsubscribe handler."""
@@ -255,7 +258,7 @@ async def test_unsubscribe_resource_delivers_uri_to_handler(connect: Connect) ->
     server = Server("library", on_unsubscribe_resource=unsubscribe_resource)
 
     async with connect(server) as client:
-        result = await client.unsubscribe_resource("file:///watched.txt")
+        result = await client.unsubscribe_resource("file:///watched.txt")  # pyright: ignore[reportDeprecated]
 
     assert result == snapshot(EmptyResult())
 

tests/interaction/lowlevel/test_subscriptions.py

@@ -0,0 +1,142 @@
+"""Client.listen stream endings against lowlevel servers over the connect matrix."""
+
+from typing import Any
+
+import anyio
+import mcp_types as types
+import pytest
+
+from mcp import MCPError
+from mcp.client.subscriptions import SubscriptionLost, ToolsListChanged
+from mcp.server import Server, ServerRequestContext
+from mcp.server.subscriptions import SUBSCRIPTION_ID_META_KEY, InMemorySubscriptionBus, ListenHandler
+from tests.interaction._connect import Connect
+from tests.interaction._requirements import requirement
+
+pytestmark = pytest.mark.anyio
+
+
+@requirement("subscriptions:listen:client:graceful-close")
+async def test_a_graceful_server_close_ends_iteration_after_buffered_events(connect: Connect) -> None:
+    """`ListenHandler.close()` sends the result last; iteration drains published events, then ends cleanly."""
+    bus = InMemorySubscriptionBus()
+    handler = ListenHandler(bus)
+    server = Server("subs", on_subscriptions_listen=handler)
+    events: list[object] = []
+    async with connect(server) as client:
+        with anyio.fail_after(10):
+            async with client.listen(tools_list_changed=True) as sub:  # pragma: no branch
+                await bus.publish(ToolsListChanged())
+                handler.close()
+                events.extend([event async for event in sub])
+    assert events == [ToolsListChanged()]
+
+
+@requirement("subscriptions:listen:client:lost")
+async def test_a_stream_dropped_after_the_ack_raises_subscription_lost(connect: Connect) -> None:
+    """Erroring the listen request after the ack (abrupt, not graceful) raises SubscriptionLost from iteration."""
+    proceed = anyio.Event()
+
+    async def dropping_listen(
+        ctx: ServerRequestContext[Any, Any], params: types.SubscriptionsListenRequestParams
+    ) -> types.SubscriptionsListenResult:
+        assert ctx.request_id is not None
+        await ctx.session.send_notification(
+            types.SubscriptionsAcknowledgedNotification(
+                params=types.SubscriptionsAcknowledgedNotificationParams(
+                    notifications=params.notifications,
+                    _meta={SUBSCRIPTION_ID_META_KEY: ctx.request_id},
+                )
+            ),
+            related_request_id=ctx.request_id,
+        )
+        await proceed.wait()
+        raise MCPError(types.INTERNAL_ERROR, "stream torn down")
+
+    server = Server("subs", on_subscriptions_listen=dropping_listen)
+    async with connect(server) as client:
+        with anyio.fail_after(10):
+            async with client.listen(tools_list_changed=True) as sub:  # pragma: no branch
+                proceed.set()
+                with pytest.raises(SubscriptionLost):  # pragma: no branch
+                    await anext(sub)
+
+
+@requirement("protocol:request-id:caller-supplied")
+async def test_the_subscription_id_is_the_listen_request_id_the_server_saw(connect: Connect) -> None:
+    """The handle's `subscription_id` is the listen request's own JSON-RPC id, known to the caller
+    while the request is still in flight - the key the server stamps every frame with for demux.
+
+    The assertion runs inside the open stream: the ack has arrived but the listen request's
+    response has not, so the id cannot have come from a response.
+    """
+    bus = InMemorySubscriptionBus()
+    stock = ListenHandler(bus)
+    seen: list[types.RequestId] = []
+
+    async def recording_listen(
+        ctx: ServerRequestContext[Any, Any], params: types.SubscriptionsListenRequestParams
+    ) -> types.SubscriptionsListenResult:
+        assert ctx.request_id is not None
+        seen.append(ctx.request_id)
+        return await stock(ctx, params)
+
+    server = Server("subs", on_subscriptions_listen=recording_listen)
+    async with connect(server) as client:
+        with anyio.fail_after(10):
+            async with client.listen(tools_list_changed=True) as sub:  # pragma: no branch
+                assert seen == [sub.subscription_id]
+                stock.close()
+                async for _event in sub:
+                    raise NotImplementedError  # unreachable: nothing was published
+
+
+@requirement("subscriptions:listen:client:concurrent-demux")
+@requirement("protocol:request-id:caller-supplied")
+async def test_concurrent_listen_streams_each_receive_their_own_ack(connect: Connect) -> None:
+    """Two subscriptions opened concurrently each surface the honored filter of their own request:
+    ack frames route by subscription id, not broadcast to every open route.
+
+    The server gates both acks until both listen requests have arrived, so both client routes are
+    live and unacknowledged when the first ack lands - a client that broadcast subscription frames
+    would cross-pollute that ack into both handles.
+    """
+    bus = InMemorySubscriptionBus()
+    stock = ListenHandler(bus)
+    arrived: list[types.RequestId] = []
+    both_arrived = anyio.Event()
+
+    async def gated_listen(
+        ctx: ServerRequestContext[Any, Any], params: types.SubscriptionsListenRequestParams
+    ) -> types.SubscriptionsListenResult:
+        assert ctx.request_id is not None
+        arrived.append(ctx.request_id)
+        if len(arrived) == 2:
+            both_arrived.set()
+        with anyio.fail_after(10):
+            await both_arrived.wait()
+        return await stock(ctx, params)
+
+    server = Server("subs", on_subscriptions_listen=gated_listen)
+    honored: dict[str, types.SubscriptionFilter] = {}
+
+    async with connect(server) as client:
+
+        async def open_tools() -> None:
+            async with client.listen(tools_list_changed=True) as sub:
+                honored["tools"] = sub.honored
+
+        async def open_prompts() -> None:
+            async with client.listen(prompts_list_changed=True) as sub:
+                honored["prompts"] = sub.honored
+
+        with anyio.fail_after(10):
+            async with anyio.create_task_group() as tg:  # pragma: no branch
+                tg.start_soon(open_tools)
+                tg.start_soon(open_prompts)
+
+    assert honored == {
+        "tools": types.SubscriptionFilter(tools_list_changed=True),
+        "prompts": types.SubscriptionFilter(prompts_list_changed=True),
+    }
+    assert len(set(arrived)) == 2

tests/interaction/mcpserver/test_subscriptions.py

@@ -0,0 +1,62 @@
+"""Client.listen against MCPServer over the connect matrix (2026-07-28)."""
+
+import anyio
+import pytest
+
+from mcp.client.subscriptions import ListenNotSupportedError, ResourceUpdated, ToolsListChanged
+from mcp.server.mcpserver import Context, MCPServer
+from tests.interaction._connect import Connect
+from tests.interaction._requirements import requirement
+
+pytestmark = pytest.mark.anyio
+
+
+def _notebook() -> MCPServer:
+    mcp = MCPServer("notebook")
+
+    @mcp.tool()
+    async def touch_tools(ctx: Context) -> str:
+        await ctx.notify_tools_changed()
+        return "ok"
+
+    @mcp.tool()
+    async def edit_note(name: str, ctx: Context) -> str:
+        await ctx.notify_resource_updated(f"note://{name}")
+        return "saved"
+
+    return mcp
+
+
+@requirement("subscriptions:listen:client:honored-surfacing")
+@requirement("subscriptions:listen:client:iteration")
+async def test_listen_surfaces_the_ack_and_iterates_typed_events(connect: Connect) -> None:
+    """Entering waits for the ack (honored is set before any event); iteration yields
+    only the typed event kinds this stream opted in to."""
+    mcp = _notebook()
+    async with connect(mcp) as client:
+        with anyio.fail_after(10):
+            async with client.listen(  # pragma: no branch
+                tools_list_changed=True, resource_subscriptions=["note://todo"]
+            ) as sub:
+                assert sub.honored.tools_list_changed is True
+                assert sub.honored.resource_subscriptions == ["note://todo"]
+
+                await client.call_tool("edit_note", {"name": "journal"})  # unsubscribed URI: silent
+                await client.call_tool("edit_note", {"name": "todo"})
+                assert await anext(sub) == ResourceUpdated(uri="note://todo")
+
+                await client.call_tool("touch_tools", {})
+                assert await anext(sub) == ToolsListChanged()
+
+
+@requirement("subscriptions:listen:client:era-guard")
+async def test_listen_on_a_pre_2026_connection_raises_the_typed_steer(connect: Connect) -> None:
+    """On 2025-era connections the guard fires before anything touches the wire, steering to the legacy verbs."""
+    mcp = _notebook()
+    async with connect(mcp) as client:
+        with anyio.fail_after(10):
+            # Entering is where the guard fires; __aenter__ directly avoids an unreachable with-body.
+            with pytest.raises(ListenNotSupportedError) as exc_info:
+                await client.listen(tools_list_changed=True).__aenter__()
+            assert exc_info.value.negotiated_version == client.session.protocol_version
+            assert "subscribe_resource" in str(exc_info.value)

tests/shared/test_dispatcher.py

@@ -25,7 +25,7 @@ from mcp_types import (
 
 from mcp.shared._compat import resync_tracer
 from mcp.shared.direct_dispatcher import DirectDispatcher, create_direct_dispatcher_pair
-from mcp.shared.dispatcher import DispatchContext, Dispatcher, OnNotify, OnRequest, Outbound
+from mcp.shared.dispatcher import DispatchContext, Dispatcher, OnNotify, OnNotifyIntercept, OnRequest, Outbound
 from mcp.shared.exceptions import MCPError
 from mcp.shared.transport_context import TransportContext
 
@@ -66,6 +66,7 @@ async def running_pair(
     server_on_notify: OnNotify | None = None,
     client_on_request: OnRequest | None = None,
     client_on_notify: OnNotify | None = None,
+    client_on_notify_intercept: OnNotifyIntercept | None = None,
     can_send_request: bool = True,
 ) -> AsyncIterator[tuple[Dispatcher[TransportContext], Dispatcher[TransportContext], Recorder, Recorder]]:
     """Yield `(client, server, client_recorder, server_recorder)` with both `run()` loops live."""
@@ -75,7 +76,9 @@ async def running_pair(
     s_req, s_notify = echo_handlers(server_rec)
     try:
         async with anyio.create_task_group() as tg:
-            await tg.start(client.run, client_on_request or c_req, client_on_notify or c_notify)
+            await tg.start(
+                client.run, client_on_request or c_req, client_on_notify or c_notify, client_on_notify_intercept
+            )
             await tg.start(server.run, server_on_request or s_req, server_on_notify or s_notify)
             try:
                 yield client, server, client_rec, server_rec
@@ -509,6 +512,67 @@ async def test_supplied_numeric_string_id_collides_with_its_int_twin(pair_factor
             assert await client.send_raw_request("again", None, {"request_id": "7"}) == {}
 
 
+@pytest.mark.anyio
+async def test_notify_intercept_sees_every_notification_and_consumes_on_true(pair_factory: PairFactory):
+    """The intercept sees every inbound notification; a frame it consumes never reaches `on_notify`, the rest do."""
+    intercepted: list[str] = []
+
+    def intercept(method: str, params: Mapping[str, Any] | None) -> bool:
+        intercepted.append(method)
+        return method == "notifications/consumed"
+
+    async with running_pair(pair_factory, client_on_notify_intercept=intercept) as (_client, server, crec, _srec):
+        with anyio.fail_after(5):
+            await server.notify("notifications/consumed", None)
+            await server.notify("notifications/passed", None)
+            await crec.notified.wait()
+    assert intercepted == ["notifications/consumed", "notifications/passed"]
+    assert [method for method, _ in crec.notifications] == ["notifications/passed"]
+
+
+@pytest.mark.anyio
+async def test_notify_intercept_completes_before_a_later_response_resolves(pair_factory: PairFactory):
+    """Notifications written before a response are intercepted before it resolves, whatever spawned handlers do."""
+    seen: list[str] = []
+
+    def intercept(method: str, params: Mapping[str, Any] | None) -> bool:
+        seen.append(method)
+        return False
+
+    async def notify_then_answer(
+        ctx: DispatchContext[TransportContext], method: str, params: Mapping[str, Any] | None
+    ) -> dict[str, Any]:
+        await ctx.notify("notifications/first", None)
+        await ctx.notify("notifications/second", None)
+        return {}
+
+    async with running_pair(
+        pair_factory, server_on_request=notify_then_answer, client_on_notify_intercept=intercept
+    ) as (client, *_):
+        with anyio.fail_after(5):
+            await client.send_raw_request("burst", None)
+        assert seen == ["notifications/first", "notifications/second"]
+
+
+@pytest.mark.anyio
+async def test_a_raising_notify_intercept_is_contained_and_passes_the_frame_through(pair_factory: PairFactory):
+    """An intercept exception costs only that interception: the frame still reaches `on_notify`, the loop survives."""
+
+    def broken_intercept(method: str, params: Mapping[str, Any] | None) -> bool:
+        raise RuntimeError("intercept exploded")
+
+    async with running_pair(pair_factory, client_on_notify_intercept=broken_intercept) as (
+        _client,
+        server,
+        crec,
+        _srec,
+    ):
+        with anyio.fail_after(5):
+            await server.notify("notifications/survives", None)
+            await crec.notified.wait()
+    assert [method for method, _ in crec.notifications] == ["notifications/survives"]
+
+
 if TYPE_CHECKING:
     _d: Dispatcher[TransportContext] = DirectDispatcher(TransportContext(kind="direct", can_send_request=True))
     _o: Outbound = _d