Kludex · Sep 16, 2026 · Sep 16, 2026 · Sep 16, 2026 · Sep 16, 2026 · Sep 16, 2026
diff --git a/docs/client/subscriptions.md b/docs/client/subscriptions.md
@@ -59,6 +59,8 @@ Requests run freely beside an open stream, from the watcher task or any other, o
 
 To stop watching, leave the block: there is no `unsubscribe` call. Cancelling the task that owns the block does that for you, and the SDK cancels the listen request the way the transport expects: over streamable HTTP, by closing that request's stream. A watcher that runs for the life of your app never returns on its own, so cancel it, or its task group's scope, at shutdown.
 
+Exit waits up to five seconds for the SDK's listen task to finish, even if the caller is cancelled. With an in-memory server, this lets cooperative handler cleanup finish before you open another subscription. The limit prevents an uncooperative handler from blocking subscription exit indefinitely. It is not an acknowledgment that a remote server has finished its cleanup.
+
 ## Streams end
 
 A stream ends in one of two ways, both ordinary control flow. A graceful server close ends the `async for`; an abrupt drop raises `SubscriptionLost`.

diff --git a/docs/get-started/testing.md b/docs/get-started/testing.md
@@ -12,18 +12,18 @@ Let's assume you have a simple server with a single tool:
 --8<-- "docs_src/testing/tutorial001.py"
 ```
 
-To run the test below you'll need two extra (development) dependencies:
+Install the development dependencies to run the test on both async backends:
 
 === "uv"
 
     ```bash
-    uv add --dev pytest inline-snapshot
+    uv add --dev pytest inline-snapshot trio
     ```
 
 === "pip"
 
     ```bash
-    pip install pytest inline-snapshot
+    pip install pytest inline-snapshot trio
     ```
 
 !!! info
@@ -45,9 +45,9 @@ from mcp.types import CallToolResult, TextContent
 from server import mcp
 
 
-@pytest.fixture
-def anyio_backend():  # (1)!
-    return "asyncio"
+@pytest.fixture(params=["asyncio", "trio"])
+def anyio_backend(request: pytest.FixtureRequest) -> str:  # (1)!
+    return request.param
 
 
 @pytest.fixture
@@ -69,7 +69,7 @@ async def test_call_add_tool(client: Client):
     )
 ```
 
-1. If you are using `trio`, return `"trio"` instead. See the [anyio documentation](https://anyio.readthedocs.io/en/stable/testing.html#specifying-the-backends-to-run-on) for the details.
+1. Each test runs once with `asyncio` and once with `trio`. Testing both catches backend-specific assumptions and scheduling races. If your application requires one backend, keep only that name in `params`. See the [anyio documentation](https://anyio.readthedocs.io/en/stable/testing.html#specifying-the-backends-to-run-on) for details.
 2. The fixture yields a connected client. Every test that takes `client` gets a fresh in-memory connection to the same server.
 
 There you go! You can now extend your tests to cover more scenarios.

diff --git a/docs/run/legacy-clients.md b/docs/run/legacy-clients.md
@@ -66,6 +66,12 @@ On one worker that is invisible. On two, it is the whole problem: a request that
     events to a client reconnecting to the *same* session), not a session store. It never makes a
     session reachable from another process.
 
+!!! note "Slow replay delays messages in the same session"
+    With `event_store=`, the SDK pauses new message delivery within the session until replay
+    finishes and the live stream is registered. This prevents responses from being lost between
+    replay and live delivery. A slow replay can delay other requests in that session, but not
+    other sessions.
+
 ## Session lifetime and limits
 
 A legacy session does not live forever, and one process does not hold an unlimited number of

diff --git a/pyproject.toml b/pyproject.toml
@@ -284,6 +284,8 @@ filterwarnings = [
     # CI and scripts/test set PYTHONWARNDEFAULTENCODING=1, so "error" rejects any text I/O
     # of ours that omits encoding=; pytest-examples' own unguarded text I/O isn't ours.
     "ignore:'encoding' argument not specified:EncodingWarning:pytest_examples",
+    # Trio wraps Linux pidfds in text mode without encoding=; only fileno()/close() are used.
+    "ignore:'encoding' argument not specified:EncodingWarning:trio\\._subprocess$",
 ]
 
 [tool.markdown.lint]

diff --git a/src/mcp/client/sse.py b/src/mcp/client/sse.py
@@ -16,6 +16,7 @@
     McpHttpClientFactory,
     create_mcp_http_client,
     request_within_origin,
+    sse_events,
     sse_within_origin,
 )
 from mcp.shared.message import SessionMessage
@@ -76,48 +77,49 @@ async def sse_client(
 
             async def sse_reader(task_status: TaskStatus[str] = anyio.TASK_STATUS_IGNORED):
                 try:
-                    async for sse in event_source:  # pragma: no branch
-                        logger.debug(f"Received SSE event: {sse.event}")
-                        match sse.event:
-                            case "endpoint":
-                                endpoint_url = urljoin(url, sse.data)
-                                logger.debug(f"Received endpoint URL: {endpoint_url}")
-
-                                url_parsed = urlparse(url)
-                                endpoint_parsed = urlparse(endpoint_url)
-                                if (  # pragma: no cover
-                                    url_parsed.netloc != endpoint_parsed.netloc
-                                    or url_parsed.scheme != endpoint_parsed.scheme
-                                ):
-                                    error_msg = (  # pragma: no cover
-                                        f"Endpoint origin does not match connection origin: {endpoint_url}"
-                                    )
-                                    logger.error(error_msg)  # pragma: no cover
-                                    raise ValueError(error_msg)  # pragma: no cover
-
-                                if on_session_created:
-                                    session_id = _extract_session_id_from_endpoint(endpoint_url)
-                                    if session_id:
-                                        on_session_created(session_id)
-
-                                task_status.started(endpoint_url)
-
-                            case "message":
-                                # Skip empty data (keep-alive pings)
-                                if not sse.data:
-                                    continue
-                                try:
-                                    message = types.jsonrpc_message_adapter.validate_json(sse.data, by_name=False)
-                                    logger.debug(f"Received server message: {message}")
-                                except Exception as exc:  # pragma: no cover
-                                    logger.exception("Error parsing server message")  # pragma: no cover
-                                    await read_stream_writer.send(exc)  # pragma: no cover
-                                    continue  # pragma: no cover
-
-                                session_message = SessionMessage(message)
-                                await read_stream_writer.send(session_message)
-                            case _:  # pragma: no cover
-                                logger.warning(f"Unknown SSE event: {sse.event}")  # pragma: no cover
+                    async with sse_events(event_source) as events:
+                        async for sse in events:  # pragma: no branch
+                            logger.debug(f"Received SSE event: {sse.event}")
+                            match sse.event:
+                                case "endpoint":
+                                    endpoint_url = urljoin(url, sse.data)
+                                    logger.debug(f"Received endpoint URL: {endpoint_url}")
+
+                                    url_parsed = urlparse(url)
+                                    endpoint_parsed = urlparse(endpoint_url)
+                                    if (  # pragma: no cover
+                                        url_parsed.netloc != endpoint_parsed.netloc
+                                        or url_parsed.scheme != endpoint_parsed.scheme
+                                    ):
+                                        error_msg = (  # pragma: no cover
+                                            f"Endpoint origin does not match connection origin: {endpoint_url}"
+                                        )
+                                        logger.error(error_msg)  # pragma: no cover
+                                        raise ValueError(error_msg)  # pragma: no cover
+
+                                    if on_session_created:
+                                        session_id = _extract_session_id_from_endpoint(endpoint_url)
+                                        if session_id:
+                                            on_session_created(session_id)
+
+                                    task_status.started(endpoint_url)
+
+                                case "message":
+                                    # Skip empty data (keep-alive pings)
+                                    if not sse.data:
+                                        continue
+                                    try:
+                                        message = types.jsonrpc_message_adapter.validate_json(sse.data, by_name=False)
+                                        logger.debug(f"Received server message: {message}")
+                                    except Exception as exc:  # pragma: no cover
+                                        logger.exception("Error parsing server message")  # pragma: no cover
+                                        await read_stream_writer.send(exc)  # pragma: no cover
+                                        continue  # pragma: no cover
+
+                                    session_message = SessionMessage(message)
+                                    await read_stream_writer.send(session_message)
+                                case _:  # pragma: no cover
+                                    logger.warning(f"Unknown SSE event: {sse.event}")  # pragma: no cover
                 except SSEError as sse_exc:  # pragma: lax no cover
                     logger.exception("Encountered SSE exception")
                     raise sse_exc

diff --git a/src/mcp/client/streamable_http.py b/src/mcp/client/streamable_http.py
@@ -37,6 +37,7 @@
     create_mcp_http_client,
     redirect_location,
     request_within_origin,
+    sse_events,
     sse_within_origin,
     stream_within_origin,
 )
@@ -231,15 +232,18 @@ async def handle_get_stream(self, client: httpx2.AsyncClient, read_stream_writer
                 if last_event_id:
                     headers[LAST_EVENT_ID] = last_event_id
 
-                async with sse_within_origin(client, self.url, headers=headers) as event_source:
+                async with (
+                    sse_within_origin(client, self.url, headers=headers) as event_source,
+                    sse_events(event_source) as events,
+                ):
                     if (redirect := _unfollowed_redirect(event_source.response)) is not None:
                         # The same GET would be redirected again, so retrying cannot help.
                         logger.warning(f"GET stream not opened: {redirect}")
                         return
                     event_source.response.raise_for_status()
                     logger.debug("GET SSE connection established")
 
-                    async for sse in event_source:
+                    async for sse in events:
                         # Track last event ID for reconnection
                         if sse.id:
                             last_event_id = sse.id
@@ -278,7 +282,10 @@ async def _handle_resumption_request(self, ctx: RequestContext) -> None:
         if isinstance(ctx.session_message.message, JSONRPCRequest):  # pragma: no branch
             original_request_id = ctx.session_message.message.id
 
-        async with sse_within_origin(ctx.client, self.url, headers=headers) as event_source:
+        async with (
+            sse_within_origin(ctx.client, self.url, headers=headers) as event_source,
+            sse_events(event_source) as events,
+        ):
             if (redirect := _unfollowed_redirect(event_source.response)) is not None:
                 logger.warning(redirect)
                 assert original_request_id is not None
@@ -289,7 +296,7 @@ async def _handle_resumption_request(self, ctx: RequestContext) -> None:
             event_source.response.raise_for_status()
             logger.debug("Resumption GET SSE connection established")
 
-            async for sse in event_source:  # pragma: no branch
+            async for sse in events:  # pragma: no branch
                 is_complete = await self._handle_sse_event(
                     sse,
                     ctx.read_stream_writer,
@@ -464,27 +471,27 @@ async def _handle_sse_response(
         original_request_id = ctx.session_message.message.id
 
         try:
-            event_source = EventSource(response)
-            async for sse in event_source:  # pragma: no branch
-                # Track last event ID for potential reconnection
-                if sse.id:
-                    last_event_id = sse.id
+            async with sse_events(EventSource(response)) as events:
+                async for sse in events:  # pragma: no branch
+                    # Track last event ID for potential reconnection
+                    if sse.id:
+                        last_event_id = sse.id
 
-                # Track retry interval from server
-                if sse.retry is not None:
-                    retry_interval_ms = sse.retry
+                    # Track retry interval from server
+                    if sse.retry is not None:
+                        retry_interval_ms = sse.retry
 
-                is_complete = await self._handle_sse_event(
-                    sse,
-                    ctx.read_stream_writer,
-                    original_request_id=original_request_id,
-                    resumption_callback=(ctx.metadata.on_resumption_token_update if ctx.metadata else None),
-                )
-                # If the SSE event indicates completion, like returning response/error
-                # break the loop
-                if is_complete:
-                    await response.aclose()
-                    return  # Normal completion, no reconnect needed
+                    is_complete = await self._handle_sse_event(
+                        sse,
+                        ctx.read_stream_writer,
+                        original_request_id=original_request_id,
+                        resumption_callback=(ctx.metadata.on_resumption_token_update if ctx.metadata else None),
+                    )
+                    # If the SSE event indicates completion, like returning response/error
+                    # break the loop
+                    if is_complete:
+                        await response.aclose()
+                        return  # Normal completion, no reconnect needed
         except Exception:
             logger.debug("SSE stream ended", exc_info=True)  # pragma: lax no cover
 
@@ -542,15 +549,18 @@ async def _handle_reconnection(
         headers[LAST_EVENT_ID] = last_event_id
 
         try:
-            async with sse_within_origin(ctx.client, self.url, headers=headers) as event_source:
+            async with (
+                sse_within_origin(ctx.client, self.url, headers=headers) as event_source,
+                sse_events(event_source) as events,
+            ):
                 event_source.response.raise_for_status()
                 logger.info("Reconnected to SSE stream")
 
                 # Track for potential further reconnection
                 reconnect_last_event_id: str = last_event_id
                 reconnect_retry_ms = retry_interval_ms
 
-                async for sse in event_source:
+                async for sse in events:
                     if sse.id:  # pragma: no branch
                         reconnect_last_event_id = sse.id
                     if sse.retry is not None:

diff --git a/src/mcp/client/subscriptions.py b/src/mcp/client/subscriptions.py
@@ -242,41 +242,50 @@ async def listen(
     opts: CallOptions = {"request_id": request_id}
     session._stamp(data, opts)  # pyright: ignore[reportPrivateUsage]
     driver_scope = anyio.CancelScope()
+    driver_done = anyio.Event()
 
     async def drive() -> None:
         # Deliberately no result timeout: the response arrives when the stream ends.
-        with driver_scope:
-            try:
-                await session._dispatcher.send_raw_request(  # pyright: ignore[reportPrivateUsage]
-                    data["method"], data.get("params"), opts
-                )
-            except MCPError as error:
-                route.settle("lost", error=error)
-                return
-            except ValueError as error:
-                # A raw request id collided with our minted listen id: fail this subscription
-                # and release the route in this same slice, so it cannot consume the raw caller's ack.
-                session._unregister_listen_route(request_id)  # pyright: ignore[reportPrivateUsage]
-                route.settle("lost", error=MCPError(types.INTERNAL_ERROR, str(error)))
-                return
-            # A result, whatever its body, is the spec's graceful close; with no prior ack
-            # it opens the subscription already closed.
-            route.set_acked(types.SubscriptionFilter())
-            route.settle("graceful")
+        try:
+            with driver_scope:
+                try:
+                    await session._dispatcher.send_raw_request(  # pyright: ignore[reportPrivateUsage]
+                        data["method"], data.get("params"), opts
+                    )
+                except MCPError as error:
+                    route.settle("lost", error=error)
+                    return
+                except ValueError as error:
+                    # A raw request id collided with our minted listen id: fail this subscription
+                    # and release the route in this same slice, so it cannot consume the raw caller's ack.
+                    session._unregister_listen_route(request_id)  # pyright: ignore[reportPrivateUsage]
+                    route.settle("lost", error=MCPError(types.INTERNAL_ERROR, str(error)))
+                    return
+                # A result, whatever its body, is the spec's graceful close; with no prior ack
+                # it opens the subscription already closed.
+                route.set_acked(types.SubscriptionFilter())
+                route.settle("graceful")
+        finally:
+            driver_done.set()
 
     # Register the demux route before the request is written so the ack cannot race it.
     route = session._register_listen_route(request_id)  # pyright: ignore[reportPrivateUsage]
     try:
         task_group.start_soon(drive)
-        with anyio.fail_after(session._session_read_timeout_seconds):  # pyright: ignore[reportPrivateUsage]
-            await route.acked.wait()
-        if route.honored is None:
-            # Only reachable on failure paths: a graceful no-ack result acked an empty filter in drive().
-            if route.error is not None:
-                raise route.error
-            raise SubscriptionLost(f"subscription {request_id!r} ended before it was acknowledged")
-        yield Subscription(route, request_id, route.honored, on_event)
+        try:
+            with anyio.fail_after(session._session_read_timeout_seconds):  # pyright: ignore[reportPrivateUsage]
+                await route.acked.wait()
+            if route.honored is None:
+                # Only reachable on failure paths: a graceful no-ack result acked an empty filter in drive().
+                if route.error is not None:
+                    raise route.error
+                raise SubscriptionLost(f"subscription {request_id!r} ended before it was acknowledged")
+            yield Subscription(route, request_id, route.honored, on_event)
+        finally:
+            route.settle("local")
+            driver_scope.cancel()
+            # Direct handlers unwind in the driver; remote cancellation has no acknowledgment.
+            with anyio.move_on_after(5, shield=True):
+                await driver_done.wait()
     finally:
-        route.settle("local")
-        driver_scope.cancel()
         session._unregister_listen_route(request_id)  # pyright: ignore[reportPrivateUsage]