Restore Trio support and test both AnyIO backends

docs/client/subscriptions.md

@@ -59,6 +59,10 @@ 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.
 
+With a direct in-memory connection (`Client(server)` or a `ClientSession` using `DirectDispatcher`), exit waits up to five seconds for the listen task to finish, even if the caller is cancelled. This lets cooperative handler cleanup release its subscription slot before you open another subscription. The limit prevents an uncooperative handler from blocking exit indefinitely.
+
+With a stream-backed connection, exit cancels the listen task without waiting for its courtesy cancellation write. The session still owns that task and its cleanup. A slow transport does not delay each subscription's exit, and exit does not acknowledge that the 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`.

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.

docs/run/legacy-clients.md

@@ -66,6 +66,14 @@ 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 "Replay is buffered before network delivery"
+    With `event_store=`, the SDK collects replayed events before sending them, so a slow replay
+    reader does not hold the event-store lock. The buffer spills to a temporary file above
+    1 MiB instead of retaining the whole history in memory. Historical events, any new resumption
+    cursor, and live events are sent in that order. The lock does not serialize incoming POSTs
+    or reserve JSON-RPC request IDs. Live-stream backpressure can still delay other messages in
+    the same session.
+
 ## Session lifetime and limits
 
 A legacy session does not live forever, and one process does not hold an unlimited number of

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]

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

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: