Possible ressource leak / race condition in streamable_http_client

[h-filzer]

Initial Checks

• I confirm that I'm using the latest version of MCP Python SDK
• I confirm that I searched for my issue in https://github.com/modelcontextprotocol/python-sdk/issues before opening this issue

Description

Observed Issue

When using the MCP SDK with the streamable-http transport, the cpu usage spikes and never goes back down after sending multiple requests and exiting the client context. I observed this bahavior especially in the google adk-python when the mcp-toolset tries to close the client.

When running the testcode to reproduce i get:

Client-Side Testcode shows Thread Leaks

Iteration 1: 1 thread  → 2 threads
Iteration 2: 2 threads → 7 threads
Iteration 3: 7 threads → 7 threads
Final: 7 threads (6 leaked asyncio_* threads)

Server-Side: Exceptions

Session crashed: unhandled errors in a TaskGroup (1 sub-exception)

ClosedResourceError at session.py:349
  → _write_stream.send() fails - stream already closed

BrokenResourceError at streamable_http.py:638
  → SSE writer has no receiver

Possible Cause

1. Client sends multiple requests via write_stream.send()
2. Client exits async with streamable_http_client context
3. tg.cancel_scope.cancel() is called during cleanup
4. Server is still processing requests (e.g., slow_echo with delay)
5. Server tries to send responses via _write_stream.send()
6. Stream is already closed → ClosedResourceError
7. Background asyncio threads handling responses become orphaned
8. Threads never terminate → memory/resource leak

Expected Behavior

• No thread leaks after exiting streamable_http_client context
• Graceful handling of client disconnection on server side

Impact

• Memory leak from accumulated threads
• Resource exhaustion in long-running applications

Affected Code Paths

• mcp/client/streamable_http.py - streamable_http_client() cleanup
• mcp/shared/session.py:237-238 - cancel_scope.cancel() and __aexit__
• mcp/server/streamable_http.py:638 - SSE response handling

Example Code

"""
This script demonstrates a possible bug in the MCP SDK where cleanup of
streamable_http_client causes BrokenResourceError and leaks threads.

Bug?: When tg.cancel_scope.cancel() is called during cleanup, child tasks
spawned by post_writer (like _handle_json_response) are still trying to
use read_stream_writer. The stream gets closed before they finish,
causing BrokenResourceError and preventing proper httpx client cleanup.
"""

import asyncio
import threading
import httpx
from mcp.client.streamable_http import streamable_http_client
from mcp.shared.message import SessionMessage
from mcp.types import JSONRPCRequest, JSONRPCMessage


async def make_mcp_request(url: str, trigger_race: bool = False) -> None:
    """Make an MCP request and observe cleanup behavior."""

    print(f"[Before] Active threads: {threading.active_count()}")
    print(f"[Before] Thread names: {[t.name for t in threading.enumerate()]}")

    # Create custom httpx client with short timeouts
    http_client = httpx.AsyncClient(
        headers={"Authorization": "test"},
        timeout=httpx.Timeout(5.0, read=10.0),
    )

    try:
        async with streamable_http_client(
            url=url,
            http_client=http_client,
        ) as (read_stream, write_stream, get_session_id):
            print(f"[Connected] Session ID: {get_session_id()}")
            print(f"[Connected] Active threads: {threading.active_count()}")

            # Send an initialize request
            init_request = JSONRPCRequest(
                jsonrpc="2.0",
                id="1",
                method="initialize",
                params={
                    "protocolVersion": "2024-11-05",
                    "capabilities": {},
                    "clientInfo": {"name": "test-client", "version": "1.0.0"},
                },
            )
            await write_stream.send(SessionMessage(JSONRPCMessage(init_request)))
            print("[Sent] Initialize request")

            # Read the response
            async for message in read_stream:
                print(f"[Received] {type(message).__name__}")
                if isinstance(message, Exception):
                    print(f"[Error in stream] {message}")
                break

            if trigger_race:
                # Send multiple requests quickly to create race during cleanup
                for i in range(5):
                    tool_request = JSONRPCRequest(
                        jsonrpc="2.0",
                        id=f"tool-{i}",
                        method="tools/call",
                        params={
                            "name": "slow_echo",
                            "arguments": {"message": f"test-{i}"},
                        },
                    )
                    await write_stream.send(SessionMessage(JSONRPCMessage(tool_request)))
                print("[Sent] 5 tool requests - exiting immediately to trigger race")
                # Exit immediately without reading responses - this should trigger the race

    except Exception as e:
        import traceback
        print(f"[Error] {type(e).__name__}: {e}")
        traceback.print_exc()

    print(f"[After cleanup] Active threads: {threading.active_count()}")
    print(f"[After cleanup] Thread names: {[t.name for t in threading.enumerate()]}")

    # Wait and check if threads persist
    await asyncio.sleep(5)
    print(f"[After 5s] Active threads: {threading.active_count()}")
    print(f"[After 5s] Thread names: {[t.name for t in threading.enumerate()]}")


async def main():
    # Replace with your MCP server URL
    MCP_URL = "http://localhost:8000/mcp"

    print("=" * 60)
    print("MCP SDK Cleanup Race Condition Reproduction")
    print("=" * 60)

    # Run multiple iterations to accumulate leaked threads
    for i in range(3):
        print(f"\n--- Iteration {i + 1} ---")
        # Trigger race condition on iterations 2 and 3
        await make_mcp_request(MCP_URL, trigger_race=(i > 0))
        await asyncio.sleep(2)

    print("\n" + "=" * 60)
    print("Final State")
    print("=" * 60)
    print(f"Active threads: {threading.active_count()}")
    for t in threading.enumerate():
        print(f"  - {t.name} (daemon={t.daemon})")


if __name__ == "__main__":
    asyncio.run(main())


-----

"""Minimal MCP server for race condition testing."""

from mcp.server.fastmcp import FastMCP

mcp = FastMCP("Test Server")


@mcp.tool()
def echo(message: str) -> str:
    """Echo the message back."""
    return f"Echo: {message}"


@mcp.tool()
def slow_echo(message: str) -> str:
    """Echo with a delay to simulate work."""
    import time

    time.sleep(0.5)
    return f"Slow Echo: {message}"


if __name__ == "__main__":
    mcp.run(transport="streamable-http")

Python & MCP Python SDK

- Python: 3.13
- MCP SDK: 1.25.0
- Transport: streamable-http

[Brid9e]

I'm experiencing the exact same issue: when connections fail or I manually shut down the streamable_http service, the CPU usage spikes to 100% instantly.

[ivanbelenky]

Just did a couple of tests here,

Background asyncio threads handling responses become orphaned

Not really as far as Im concerned, the asyncio threads are due to shutdown when the event loop ends exits. What is happening here is the following

• when you send a request, getaddrinfo gets called as part of the connect_tcp call in httpcore.
• getaddrinfo as any other blocking syscall library function that hosts many blocking syscalls, gets executed on its own thread (lets say it gets called at $t_0$)
• this threads have no idle timeout or anything
• asyncio default Executor is executor = concurrent.futures.ThreadPoolExecutor(thread_name_prefix='asyncio') (you can find that line in the cpython stdlib here). This means that the max_workers setting is set by default to the maximum, usually 2 times the cores you got.
• executing getaddrinfo takes approximately $\delta t$
• (HERE is where stuff happens) if any other call to getaddrinfo happens during the range $[t_0, t_0+\delta t]$ the executor will see no worker available and will spin a new thread.

If you modify your codebase slightly and introduce a minor delay after the send you will get no more threads, just the one that got started since the first call. Furthermore, if you want to constrain this further, you can always run something like

loop.set_default_executor(concurrent.futures.ThreadPoolExecutor(max_workers=2)) # or whatever number rocks your boat

Threads should be spending 0 cpu cycles @Brid9e, would you mind providing something for reproducibility?

[challenger71498]

Hi @h-filzer ,

I also had same issue you've encountered, and find out the BaseSession cleanup logic is being cancelled on __aexit__, causing respond streams never finish.

I think your issue is could be fixed by this PR #1817, you may look into it if you are interested in.

[ivanbelenky]

@challenger71498 awesome fix, thx for it, I was able to replicate the issue above by @h-filzer rebasing locally to your fork, so I think your PR has no effect on this. I still believe this is happening because of what was explained here.

[challenger71498]

@ivanbelenky Yup, seems like this issue is not related to the my PR.

I will remove this issue from the PR. Thx 👍