Error while serving claude caude sdk over FastAPI EventStream #4
Description
I am trying to use Claude code over FastAPI SSE but getting this error.
Error:
INFO: 127.0.0.1:58739 - "POST /chat/stream HTTP/1.1" 200 OK
2025-06-13 18:53:02,174 - asyncio - ERROR - Task exception was never retrieved
future: <Task finished name='Task-12' coro=<<async_generator_athrow without __name__>()> exception=RuntimeError('Attempted to exit cancel scope in a different task than it was entered in')>
+ Exception Group Traceback (most recent call last):
| File "/Users/mohit/code/code-gen/.venv/lib/python3.13/site-packages/anyio/_backends/_asyncio.py", line 772, in __aexit__
| raise BaseExceptionGroup(
| "unhandled errors in a TaskGroup", self._exceptions
| ) from None
| BaseExceptionGroup: unhandled errors in a TaskGroup (1 sub-exception)
+-+---------------- 1 ----------------
| Traceback (most recent call last):
| File "/Users/mohit/code/code-gen/.venv/lib/python3.13/site-packages/claude_code_sdk/_internal/transport/subprocess_cli.py", line 193, in receive_messages
| yield data
| GeneratorExit
+------------------------------------
During handling of the above exception, another exception occurred:
Traceback (most recent call last):
File "/Users/mohit/code/code-gen/.venv/lib/python3.13/site-packages/claude_code_sdk/_internal/transport/subprocess_cli.py", line 182, in receive_messages
async with anyio.create_task_group() as tg:
~~~~~~~~~~~~~~~~~~~~~~~^^
File "/Users/mohit/code/code-gen/.venv/lib/python3.13/site-packages/anyio/_backends/_asyncio.py", line 778, in __aexit__
if self.cancel_scope.__exit__(type(exc), exc, exc.__traceback__):
~~~~~~~~~~~~~~~~~~~~~~~~~~^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "/Users/mohit/code/code-gen/.venv/lib/python3.13/site-packages/anyio/_backends/_asyncio.py", line 457, in __exit__
raise RuntimeError(
...<2 lines>...
)
RuntimeError: Attempted to exit cancel scope in a different task than it was entered in
Code:
async def generate():
try:
logger.info("Starting Claude Code query")
options = ClaudeCodeOptions(
max_turns=3,
system_prompt="You are a helpful assistant",
allowed_tools=["Read", "Write", "Bash"],
permission_mode="acceptEdits",
resume=request.resume if request.resume and request.continue_conversation else None,
continue_conversation=request.continue_conversation,
cwd="/Users/mohit/code/gemfire-migration/gemfire-migration/codebases/gemfire-java"
)
message_count = 0
async for message in query(prompt=request.message, options=options):
message_count += 1
if isinstance(message, AssistantMessage):
logger.debug(f"Processing AssistantMessage {message_count} with {len(message.content)} blocks")
for block in message.content:
if isinstance(block, TextBlock):
data = {"type": "content", "content": block.text}
yield f"data: {json.dumps(data)}\n\n"
elif isinstance(message, SystemMessage):
logger.debug(f"Processing SystemMessage {message_count} - Subtype: {message.subtype}")
data = {"type": "system", "subtype": message.subtype, "data": message.data}
yield f"data: {json.dumps(data)}\n\n"
elif isinstance(message, ResultMessage):
logger.info(f"Query completed - Cost: ${message.cost_usd:.4f}, Duration: {message.duration_ms}ms, Turns: {message.num_turns}, Error: {message.is_error}")
data = {
"type": "result",
"cost_usd": message.cost_usd,
"duration_ms": message.duration_ms,
"num_turns": message.num_turns,
"session_id": message.session_id,
"is_error": message.is_error
}
yield f"data: {json.dumps(data)}\n\n"
logger.info(f"Stream completed successfully - Processed {message_count} messages")
except Exception as e:
logger.error(f"Error in streaming: {str(e)}", exc_info=True)
error_data = {"type": "error", "error": str(e)}
yield f"data: {json.dumps(error_data)}\n\n"
finally:
# Send completion marker
logger.debug("Sending completion marker")
yield f"data: {json.dumps({'type': 'done'})}\n\n"
return StreamingResponse(generate(), media_type="text/event-stream")