fix: close all memory stream ends in client transport cleanup

Client transports for SSE, WebSocket, and StreamableHTTP create 4 memory
stream ends (2 paired streams) but only closed 2 in their finally blocks.
anyio memory stream ends are independent — closing the writer does not
close the reader. The unclosed ends leak and emit ResourceWarning when
garbage collected.

This caused flaky test failures in CI: a transport connection error in
one test would leak streams, then GC in a later unrelated test would
trigger ResourceWarning, which pytest promotes to a test failure.

Fix follows the existing correct pattern in stdio.py:
- sse.py: close all 4 stream ends in the existing finally block
- streamable_http.py: close all 4 stream ends in the existing finally
  block (read_stream was previously never closed, even on happy path)
- websocket.py: add try/finally wrapping the entire body, closing all
  4 stream ends (previously had no cleanup at all — ws_connect failure
  leaked everything)

Regression tests force gc.collect() after the transport context exits
so leaked streams fail deterministically in the test that caused them.

Author: maxisbey

src/mcp/client/sse.py

@@ -160,3 +160,5 @@ async def post_writer(endpoint_url: str):
         finally:
             await read_stream_writer.aclose()
             await write_stream.aclose()
+            await read_stream.aclose()
+            await write_stream_reader.aclose()

src/mcp/client/streamable_http.py

@@ -577,3 +577,5 @@ def start_get_stream() -> None:
         finally:
             await read_stream_writer.aclose()
             await write_stream.aclose()
+            await read_stream.aclose()
+            await write_stream_reader.aclose()

src/mcp/client/websocket.py

@@ -41,40 +41,46 @@ async def websocket_client(
     read_stream_writer, read_stream = anyio.create_memory_object_stream(0)
     write_stream, write_stream_reader = anyio.create_memory_object_stream(0)
 
-    # Connect using websockets, requesting the "mcp" subprotocol
-    async with ws_connect(url, subprotocols=[Subprotocol("mcp")]) as ws:
+    try:
+        # Connect using websockets, requesting the "mcp" subprotocol
+        async with ws_connect(url, subprotocols=[Subprotocol("mcp")]) as ws:
 
-        async def ws_reader():
-            """Reads text messages from the WebSocket, parses them as JSON-RPC messages,
-            and sends them into read_stream_writer.
-            """
-            async with read_stream_writer:
-                async for raw_text in ws:
-                    try:
-                        message = types.jsonrpc_message_adapter.validate_json(raw_text, by_name=False)
-                        session_message = SessionMessage(message)
-                        await read_stream_writer.send(session_message)
-                    except ValidationError as exc:  # pragma: no cover
-                        # If JSON parse or model validation fails, send the exception
-                        await read_stream_writer.send(exc)
+            async def ws_reader():
+                """Reads text messages from the WebSocket, parses them as JSON-RPC messages,
+                and sends them into read_stream_writer.
+                """
+                async with read_stream_writer:
+                    async for raw_text in ws:
+                        try:
+                            message = types.jsonrpc_message_adapter.validate_json(raw_text, by_name=False)
+                            session_message = SessionMessage(message)
+                            await read_stream_writer.send(session_message)
+                        except ValidationError as exc:  # pragma: no cover
+                            # If JSON parse or model validation fails, send the exception
+                            await read_stream_writer.send(exc)
 
-        async def ws_writer():
-            """Reads JSON-RPC messages from write_stream_reader and
-            sends them to the server.
-            """
-            async with write_stream_reader:
-                async for session_message in write_stream_reader:
-                    # Convert to a dict, then to JSON
-                    msg_dict = session_message.message.model_dump(by_alias=True, mode="json", exclude_unset=True)
-                    await ws.send(json.dumps(msg_dict))
+            async def ws_writer():
+                """Reads JSON-RPC messages from write_stream_reader and
+                sends them to the server.
+                """
+                async with write_stream_reader:
+                    async for session_message in write_stream_reader:
+                        # Convert to a dict, then to JSON
+                        msg_dict = session_message.message.model_dump(by_alias=True, mode="json", exclude_unset=True)
+                        await ws.send(json.dumps(msg_dict))
 
-        async with anyio.create_task_group() as tg:
-            # Start reader and writer tasks
-            tg.start_soon(ws_reader)
-            tg.start_soon(ws_writer)
+            async with anyio.create_task_group() as tg:
+                # Start reader and writer tasks
+                tg.start_soon(ws_reader)
+                tg.start_soon(ws_writer)
 
-            # Yield the receive/send streams
-            yield (read_stream, write_stream)
+                # Yield the receive/send streams
+                yield (read_stream, write_stream)
 
-            # Once the caller's 'async with' block exits, we shut down
-            tg.cancel_scope.cancel()
+                # Once the caller's 'async with' block exits, we shut down
+                tg.cancel_scope.cancel()
+    finally:
+        await read_stream_writer.aclose()
+        await write_stream.aclose()
+        await read_stream.aclose()
+        await write_stream_reader.aclose()

tests/client/test_transport_stream_cleanup.py

@@ -0,0 +1,82 @@
+"""Regression tests for memory stream leaks in client transports.
+
+When a connection error occurs (404, 403, ConnectError), transport context
+managers must close ALL 4 memory stream ends they created. anyio memory streams
+are paired but independent — closing the writer does NOT close the reader.
+Unclosed stream ends emit ResourceWarning on GC, which pytest promotes to a
+test failure in whatever test happens to be running when GC triggers.
+
+These tests force GC after the transport context exits, so any leaked stream
+triggers a ResourceWarning immediately and deterministically here, rather than
+nondeterministically in an unrelated later test.
+"""
+
+import gc
+import socket
+
+import httpx
+import pytest
+
+from mcp.client.sse import sse_client
+from mcp.client.streamable_http import streamable_http_client
+from mcp.client.websocket import websocket_client
+
+
+def _unused_tcp_port() -> int:
+    """Return a port with no listener. Binding then closing leaves the port unbound."""
+    with socket.socket() as s:
+        s.bind(("127.0.0.1", 0))
+        return s.getsockname()[1]
+
+
+@pytest.mark.anyio
+async def test_sse_client_closes_all_streams_on_connection_error() -> None:
+    """sse_client must close all 4 stream ends when the connection fails.
+
+    Before the fix, only read_stream_writer and write_stream were closed in
+    the finally block. read_stream and write_stream_reader were leaked.
+    """
+    port = _unused_tcp_port()
+
+    # sse_client enters a task group BEFORE connecting, so anyio wraps the
+    # ConnectError from aconnect_sse in an ExceptionGroup. ExceptionGroup is
+    # an Exception subclass, so we catch broadly and verify the sub-exception.
+    with pytest.raises(Exception) as exc_info:  # noqa: B017
+        async with sse_client(f"http://127.0.0.1:{port}/sse"):
+            pytest.fail("should not reach here")  # pragma: no cover
+
+    assert exc_info.group_contains(httpx.ConnectError)
+
+    # If any stream leaked, gc.collect() triggers ResourceWarning in __del__,
+    # which pytest's filterwarnings=["error"] promotes to a test failure.
+    gc.collect()
+
+
+@pytest.mark.anyio
+async def test_streamable_http_client_closes_all_streams_on_exit() -> None:
+    """streamable_http_client must close all 4 stream ends on exit.
+
+    Before the fix, read_stream was never closed — not even on the happy path.
+    This test enters and exits the context without sending any messages, so no
+    network connection is ever attempted (streamable_http connects lazily).
+    """
+    async with streamable_http_client("http://127.0.0.1:1/mcp"):
+        pass
+
+    gc.collect()
+
+
+@pytest.mark.anyio
+async def test_websocket_client_closes_all_streams_on_connection_error() -> None:
+    """websocket_client must close all 4 stream ends when ws_connect fails.
+
+    Before the fix, there was no try/finally at all — if ws_connect raised,
+    all 4 streams were leaked.
+    """
+    port = _unused_tcp_port()
+
+    with pytest.raises(OSError):
+        async with websocket_client(f"ws://127.0.0.1:{port}/ws"):
+            pytest.fail("should not reach here")  # pragma: no cover
+
+    gc.collect()