test: close leaked SSE receive streams instead of gc-collecting them

SseServerTransport.connect_sse never closed sse_stream_reader after
EventSourceResponse returned; the in-process bridge dropped its chunk
reader when a request was cancelled before the response started. Close
both at source so the interaction suite no longer needs a gc.collect()
workaround, and pull one assert inside its async-with body to clear the
last 3.11 coverage gap.

Author: maxisbey

src/mcp/server/sse.py

@@ -179,6 +179,7 @@ async def response_wrapper(scope: Scope, receive: Receive, send: Send):
                 await EventSourceResponse(content=sse_stream_reader, data_sender_callable=sse_writer)(
                     scope, receive, send
                 )
+                await sse_stream_reader.aclose()
                 await read_stream_writer.aclose()
                 await write_stream_reader.aclose()
                 self._read_stream_writers.pop(session_id, None)

tests/interaction/_connect.py

@@ -7,8 +7,6 @@
 (session ids, SSE encoding, session management) runs with no sockets, threads, or subprocesses.
 """
 
-import gc
-import warnings
 from collections.abc import AsyncIterator, Awaitable, Callable, Iterable
 from contextlib import AbstractAsyncContextManager, asynccontextmanager
 from typing import Any, Protocol
@@ -347,26 +345,14 @@ def httpx_client_factory(
         )
 
     transport = sse_client(f"{BASE_URL}/sse", httpx_client_factory=httpx_client_factory)
-    try:
-        async with Client(
-            transport,
-            read_timeout_seconds=read_timeout_seconds,
-            sampling_callback=sampling_callback,
-            list_roots_callback=list_roots_callback,
-            logging_callback=logging_callback,
-            message_handler=message_handler,
-            client_info=client_info,
-            elicitation_callback=elicitation_callback,
-        ) as client:
-            yield client
-    finally:
-        # SseServerTransport.connect_sse never closes its sse_stream_reader (handed to
-        # sse_starlette.EventSourceResponse, which does not aclose() its content on cancel).
-        # After teardown that reader is held only by a reference cycle through the connect_sse
-        # frame and its task objects; collecting twice runs the cycle's finalizers and then
-        # frees the reader while ResourceWarning is suppressed, instead of at an arbitrary
-        # later GC under pytest's error filter. One pass suffices on 3.11+; 3.10 needs both.
-        with warnings.catch_warnings():
-            warnings.simplefilter("ignore", ResourceWarning)
-            gc.collect()
-            gc.collect()
+    async with Client(
+        transport,
+        read_timeout_seconds=read_timeout_seconds,
+        sampling_callback=sampling_callback,
+        list_roots_callback=list_roots_callback,
+        logging_callback=logging_callback,
+        message_handler=message_handler,
+        client_info=client_info,
+        elicitation_callback=elicitation_callback,
+    ) as client:
+        yield client

tests/interaction/transports/_bridge.py

@@ -151,11 +151,16 @@ async def run_application() -> None:
                 await chunk_writer.aclose()
 
         self._task_group.start_soon(run_application)
-        await response_started.wait()
-        if application_error is not None:
-            # No response will be built, so close the reader the response body would have owned.
+        try:
+            await response_started.wait()
+            if application_error is not None:
+                raise application_error
+        except BaseException:
+            # No response will be built, so close the reader the response body would have owned
+            # and tell the application its peer has gone away.
+            client_disconnected.set()
             await chunk_reader.aclose()
-            raise application_error
+            raise
         return httpx.Response(
             status_code=response_status,
             headers=response_headers,

tests/interaction/transports/test_hosting_session.py

@@ -115,17 +115,16 @@ async def test_delete_terminates_the_session_and_subsequent_requests_return_404(
             json={"jsonrpc": "2.0", "id": 2, "method": "tools/list"},
             headers=base_headers(session_id=session_id),
         )
-
-    assert (post.status_code, post.json()) == snapshot(
-        (
-            404,
-            {
-                "jsonrpc": "2.0",
-                "id": None,
-                "error": {"code": -32600, "message": "Not Found: Session has been terminated"},
-            },
+        assert (post.status_code, post.json()) == snapshot(
+            (
+                404,
+                {
+                    "jsonrpc": "2.0",
+                    "id": None,
+                    "error": {"code": -32600, "message": "Not Found: Session has been terminated"},
+                },
+            )
         )
-    )
 
 
 @requirement("hosting:session:isolation")

tests/interaction/transports/test_sse.py

@@ -7,8 +7,6 @@
 through the suite's streaming ASGI bridge.
 """
 
-import gc
-import warnings
 from uuid import UUID, uuid4
 
 import anyio
@@ -59,10 +57,6 @@ def httpx_client_factory(
             assert await client.send_ping() == snapshot(EmptyResult())
 
     assert sse._read_stream_writers == {}
-    # See connect_over_sse: collect the one stream sse_starlette never closes on disconnect.
-    with warnings.catch_warnings():
-        warnings.simplefilter("ignore", ResourceWarning)
-        gc.collect()
 
 
 @requirement("transport:sse:post:session-routing")