René's URL Explorer Experiment

{"@context":"https://schema.org","@type":"DiscussionForumPosting","headline":"Possible ressource leak / race condition in streamable_http_client","articleBody":"### Initial Checks\n\n- [x] I confirm that I'm using the latest version of MCP Python SDK\n- [x] I confirm that I searched for my issue in https://github.com/modelcontextprotocol/python-sdk/issues before opening this issue\n\n### Description\n\n## Observed Issue\nWhen 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.\n\nWhen running the testcode to reproduce i get:\n\n### Client-Side Testcode shows Thread Leaks\n```\nIteration 1: 1 thread  → 2 threads\nIteration 2: 2 threads → 7 threads\nIteration 3: 7 threads → 7 threads\nFinal: 7 threads (6 leaked asyncio_* threads)\n```\n\n### Server-Side: Exceptions\n```\nSession crashed: unhandled errors in a TaskGroup (1 sub-exception)\n\nClosedResourceError at session.py:349\n  → _write_stream.send() fails - stream already closed\n\nBrokenResourceError at streamable_http.py:638\n  → SSE writer has no receiver\n```\n\n## Possible Cause\n\n1. Client sends multiple requests via `write_stream.send()`\n2. Client exits `async with streamable_http_client` context\n3. `tg.cancel_scope.cancel()` is called during cleanup\n4. Server is still processing requests (e.g., `slow_echo` with delay)\n5. Server tries to send responses via `_write_stream.send()`\n6. Stream is already closed → `ClosedResourceError`\n7. Background asyncio threads handling responses become orphaned\n8. Threads never terminate → memory/resource leak\n\n\n## Expected Behavior\n- No thread leaks after exiting `streamable_http_client` context\n- Graceful handling of client disconnection on server side\n\n## Impact\n- Memory leak from accumulated threads\n- Resource exhaustion in long-running applications\n\n## Affected Code Paths\n- `mcp/client/streamable_http.py` - `streamable_http_client()` cleanup\n- `mcp/shared/session.py:237-238` - `cancel_scope.cancel()` and `__aexit__`\n- `mcp/server/streamable_http.py:638` - SSE response handling\n\n\n\n### Example Code\n\n```Python\n\"\"\"\nThis script demonstrates a possible bug in the MCP SDK where cleanup of\nstreamable_http_client causes BrokenResourceError and leaks threads.\n\nBug?: When tg.cancel_scope.cancel() is called during cleanup, child tasks\nspawned by post_writer (like _handle_json_response) are still trying to\nuse read_stream_writer. The stream gets closed before they finish,\ncausing BrokenResourceError and preventing proper httpx client cleanup.\n\"\"\"\n\nimport asyncio\nimport threading\nimport httpx\nfrom mcp.client.streamable_http import streamable_http_client\nfrom mcp.shared.message import SessionMessage\nfrom mcp.types import JSONRPCRequest, JSONRPCMessage\n\n\nasync def make_mcp_request(url: str, trigger_race: bool = False) -\u003e None:\n    \"\"\"Make an MCP request and observe cleanup behavior.\"\"\"\n\n    print(f\"[Before] Active threads: {threading.active_count()}\")\n    print(f\"[Before] Thread names: {[t.name for t in threading.enumerate()]}\")\n\n    # Create custom httpx client with short timeouts\n    http_client = httpx.AsyncClient(\n        headers={\"Authorization\": \"test\"},\n        timeout=httpx.Timeout(5.0, read=10.0),\n    )\n\n    try:\n        async with streamable_http_client(\n            url=url,\n            http_client=http_client,\n        ) as (read_stream, write_stream, get_session_id):\n            print(f\"[Connected] Session ID: {get_session_id()}\")\n            print(f\"[Connected] Active threads: {threading.active_count()}\")\n\n            # Send an initialize request\n            init_request = JSONRPCRequest(\n                jsonrpc=\"2.0\",\n                id=\"1\",\n                method=\"initialize\",\n                params={\n                    \"protocolVersion\": \"2024-11-05\",\n                    \"capabilities\": {},\n                    \"clientInfo\": {\"name\": \"test-client\", \"version\": \"1.0.0\"},\n                },\n            )\n            await write_stream.send(SessionMessage(JSONRPCMessage(init_request)))\n            print(\"[Sent] Initialize request\")\n\n            # Read the response\n            async for message in read_stream:\n                print(f\"[Received] {type(message).__name__}\")\n                if isinstance(message, Exception):\n                    print(f\"[Error in stream] {message}\")\n                break\n\n            if trigger_race:\n                # Send multiple requests quickly to create race during cleanup\n                for i in range(5):\n                    tool_request = JSONRPCRequest(\n                        jsonrpc=\"2.0\",\n                        id=f\"tool-{i}\",\n                        method=\"tools/call\",\n                        params={\n                            \"name\": \"slow_echo\",\n                            \"arguments\": {\"message\": f\"test-{i}\"},\n                        },\n                    )\n                    await write_stream.send(SessionMessage(JSONRPCMessage(tool_request)))\n                print(\"[Sent] 5 tool requests - exiting immediately to trigger race\")\n                # Exit immediately without reading responses - this should trigger the race\n\n    except Exception as e:\n        import traceback\n        print(f\"[Error] {type(e).__name__}: {e}\")\n        traceback.print_exc()\n\n    print(f\"[After cleanup] Active threads: {threading.active_count()}\")\n    print(f\"[After cleanup] Thread names: {[t.name for t in threading.enumerate()]}\")\n\n    # Wait and check if threads persist\n    await asyncio.sleep(5)\n    print(f\"[After 5s] Active threads: {threading.active_count()}\")\n    print(f\"[After 5s] Thread names: {[t.name for t in threading.enumerate()]}\")\n\n\nasync def main():\n    # Replace with your MCP server URL\n    MCP_URL = \"http://localhost:8000/mcp\"\n\n    print(\"=\" * 60)\n    print(\"MCP SDK Cleanup Race Condition Reproduction\")\n    print(\"=\" * 60)\n\n    # Run multiple iterations to accumulate leaked threads\n    for i in range(3):\n        print(f\"\\n--- Iteration {i + 1} ---\")\n        # Trigger race condition on iterations 2 and 3\n        await make_mcp_request(MCP_URL, trigger_race=(i \u003e 0))\n        await asyncio.sleep(2)\n\n    print(\"\\n\" + \"=\" * 60)\n    print(\"Final State\")\n    print(\"=\" * 60)\n    print(f\"Active threads: {threading.active_count()}\")\n    for t in threading.enumerate():\n        print(f\"  - {t.name} (daemon={t.daemon})\")\n\n\nif __name__ == \"__main__\":\n    asyncio.run(main())\n\n\n-----\n\n\"\"\"Minimal MCP server for race condition testing.\"\"\"\n\nfrom mcp.server.fastmcp import FastMCP\n\nmcp = FastMCP(\"Test Server\")\n\n\n@mcp.tool()\ndef echo(message: str) -\u003e str:\n    \"\"\"Echo the message back.\"\"\"\n    return f\"Echo: {message}\"\n\n\n@mcp.tool()\ndef slow_echo(message: str) -\u003e str:\n    \"\"\"Echo with a delay to simulate work.\"\"\"\n    import time\n\n    time.sleep(0.5)\n    return f\"Slow Echo: {message}\"\n\n\nif __name__ == \"__main__\":\n    mcp.run(transport=\"streamable-http\")\n```\n\n### Python \u0026 MCP Python SDK\n\n```Text\n- Python: 3.13\n- MCP SDK: 1.25.0\n- Transport: streamable-http\n```","author":{"url":"https://github.com/h-filzer","@type":"Person","name":"h-filzer"},"datePublished":"2025-12-19T13:46:48.000Z","interactionStatistic":{"@type":"InteractionCounter","interactionType":"https://schema.org/CommentAction","userInteractionCount":6},"url":"https://github.com/1805/python-sdk/issues/1805"}