fix: clean up StreamableHTTP SSE disconnects

[tarunag10]

Summary

• Ensure POST SSE responses clean up their per-request SSE writer and memory streams when the response returns.
• Add a regression test for the response-return/disconnect path so per-request stream registries do not retain stale entries.

Why

Issue #2958 reports StreamableHTTP servers accumulating CLOSE_WAIT sockets behind reverse proxies. One failure mode is that EventSourceResponse can return after a client/proxy disconnect while the per-request writer remains blocked on the in-memory request stream. Closing the request streams in the response task's finally path unblocks the writer and lets the ASGI request finish cleanly.

Validation

• uv run ruff format --check src/mcp/server/streamable_http.py tests/server/test_streamable_http_manager.py
• uv run ruff check src/mcp/server/streamable_http.py tests/server/test_streamable_http_manager.py
• uv run pytest tests/server/test_streamable_http_manager.py -q

Fixes #2958

[cubic-dev-ai]

1 issue found across 2 files

Prompt for AI agents (unresolved issues)

Check if these issues are valid — if so, understand the root cause of each and fix them. If appropriate, use sub-agents to investigate and fix each issue separately.


<file name="tests/server/test_streamable_http_manager.py">

<violation number="1" location="tests/server/test_streamable_http_manager.py:138">
P2: The new async test uses an unbounded `read_stream.receive()` wait, which can hang the test suite on failure paths instead of timing out deterministically.</violation>
</file>

<sub>Reply with feedback, questions, or to request a fix.

Fix all with cubic | Re-trigger cubic</sub>

[cubic-dev-ai]

P2: The new async test uses an unbounded read_stream.receive() wait, which can hang the test suite on failure paths instead of timing out deterministically.

Prompt for AI agents

Check if this issue is valid — if so, understand the root cause and fix it. At tests/server/test_streamable_http_manager.py, line 138:

<comment>The new async test uses an unbounded `read_stream.receive()` wait, which can hang the test suite on failure paths instead of timing out deterministically.</comment>

<file context>
@@ -101,6 +101,48 @@ async def running_manager():
+    async with transport.connect() as (read_stream, _write_stream):
+        async with anyio.create_task_group() as tg:
+            tg.start_soon(transport.handle_request, scope, receive, send)
+            session_message = await read_stream.receive()
+            assert session_message.message.method == "tools/list"
+
</file context>

Suggested change

	session_message = await read_stream.receive()
	assert session_message.message.method == "tools/list"
	with anyio.fail_after(5):
	session_message = await read_stream.receive()
	assert session_message.message.method == "tools/list"

[itxaiohanglover]

Good fix! The cleanup in finally ensures SSE stream writers and memory streams are properly cleaned up when the response returns, preventing CLOSE_WAIT accumulation. The DisconnectingEventSourceResponse test mock is a clever way to verify cleanup behavior.