ClaudeSDKClient silently hangs when reused across ASGI request tasks (FastAPI/Starlette)

[naga-k]

Summary

ClaudeSDKClient silently hangs on the second receive_response() call when the client is reused across different ASGI request tasks in FastAPI/Starlette. The subprocess is alive, all internal state looks healthy, but messages never arrive. No error is raised — it just hangs forever.

Context

The Python SDK reference recommends ClaudeSDKClient for "Chat interfaces, REPLs" and "Continuous conversations." The hosting guide describes "Long-Running Sessions" and "High-Frequency Chat Bots" as deployment patterns. These are server framework use cases, but there is no guidance on how to use ClaudeSDKClient with FastAPI, Starlette, Django, or any ASGI framework.

The caveat in client.py:46-52 documents the limitation in source code, but it doesn't appear in any docs page, and the failure mode is a silent hang rather than an error — making it extremely difficult to diagnose.

Reproduction

# FastAPI app — this hangs on the second request
from fastapi import FastAPI
from claude_agent_sdk import ClaudeSDKClient, ClaudeAgentOptions

app = FastAPI()
_clients: dict[str, ClaudeSDKClient] = {}

async def get_client(session_id: str) -> ClaudeSDKClient:
    if session_id not in _clients:
        client = ClaudeSDKClient(options=ClaudeAgentOptions(...))
        await client.connect()  # Connected in ASGI task A
        _clients[session_id] = client
    return _clients[session_id]

@app.post("/chat")
async def chat(session_id: str, message: str):
    client = await get_client(session_id)
    await client.query(message, session_id=session_id)
    # Request 1: works (same ASGI task as connect())
    # Request 2: hangs forever (different ASGI task)
    async for msg in client.receive_response():
        yield msg

Same pattern works perfectly outside FastAPI:

async def main():
    client = ClaudeSDKClient(options=ClaudeAgentOptions(...))
    await client.connect()
    
    await client.query("My name is Test")
    async for msg in client.receive_response():
        pass  # Works
    
    await client.query("What's my name?")
    async for msg in client.receive_response():
        pass  # Also works — same asyncio task
        
asyncio.run(main())

Root cause

FastAPI/uvicorn handles each HTTP request in a separate asyncio task. The SDK's internal anyio task group (created during connect()) can't deliver messages to receive_response() when called from a different task. The subprocess responds, but the anyio MemoryObjectStream doesn't bridge across task boundaries.

Diagnostic evidence:

• Subprocess alive (returncode=None, pid active)
• Transport ready, stdin writable, task group not cancelled
• buffer_used=0 after waiting 2s (messages never arrive)
• Same client + options works perfectly in a single-task script

Workaround

We solved this with a dedicated asyncio.Task per session that owns the SDK client, with asyncio.Queue bridges to HTTP handlers:

class _SessionWorker:
    """Dedicated asyncio.Task per session — all SDK ops in one task context."""
    
    async def _run(self):
        client = ClaudeSDKClient(options=_build_options())
        await client.connect()  # Task W
        while True:
            message, session_id, out_q = await self._input.get()
            await client.query(message, session_id=session_id)  # Task W
            async for msg in client.receive_response():  # Task W — works!
                await out_q.put(msg)

    async def query_and_stream(self, message, session_id):
        out_q = asyncio.Queue()
        await self._input.put((message, session_id, out_q))
        while True:
            msg = await out_q.get()  # HTTP handler reads from queue
            yield msg

This works but shouldn't be necessary — the SDK should handle cross-task usage transparently, or at minimum raise an error instead of silently hanging.

Suggestions

1. Fix the limitation — make ClaudeSDKClient safe to use across asyncio tasks (the docstring already says "Ideally, this limitation should not exist")
2. Or raise an error — detect cross-task usage and raise ClaudeSDKError("ClaudeSDKClient cannot be used across different asyncio tasks...") instead of silently hanging
3. Document server framework patterns — the hosting guide and Python reference should show how to use ClaudeSDKClient in FastAPI/Starlette/Django with the worker pattern
4. Add an example — examples/fastapi_server.py showing session-scoped client reuse with the queue bridge pattern

Environment

• claude-agent-sdk v0.1.36
• Python 3.12
• FastAPI 0.115+
• uvicorn (ASGI)
• Linux