SDK MCP Servers Fail with Streaming Input in Python SDK v0.1.5

[oranshayer]

SDK MCP Servers Fail with Streaming Input in Python SDK v0.1.5

Environment

• SDK Version: claude-agent-sdk 0.1.5 (Python)
• Python Version: 3.11.9
• OS: macOS (Darwin 24.5.0)
• Claude Code CLI: @anthropic-ai/claude-code@2.0.28

Description

SDK-created MCP servers (using create_sdk_mcp_server()) fail when used with streaming input mode (async generators), despite documentation stating "Custom MCP tools require streaming input mode."

Error

CLIConnectionError: ProcessTransport is not ready for writing
Error: Expected message type 'user' or 'control', got 'undefined'

Full traceback:

ExceptionGroup: unhandled errors in a TaskGroup (1 sub-exception)
+-+---------------- 1 ----------------
  | Traceback (most recent call last):
  |   File "claude_agent_sdk/_internal/query.py", line 303, in _handle_control_request
  |     await self.transport.write(json.dumps(success_response) + "\n")
  |   File "claude_agent_sdk/_internal/transport/subprocess_cli.py", line 410, in write
  |     raise CLIConnectionError("ProcessTransport is not ready for writing")
  | claude_agent_sdk._errors.CLIConnectionError: ProcessTransport is not ready for writing

Minimal Reproduction

import asyncio
from claude_agent_sdk import query, tool, create_sdk_mcp_server
from claude_agent_sdk.types import ClaudeAgentOptions

# Define a simple tool
@tool("calculate", "Perform calculations", {"expression": str})
async def calculate(args: dict) -> dict:
    result = eval(args["expression"])
    return {"content": [{"type": "text", "text": f"Result: {result}"}]}

# Create SDK MCP server
server = create_sdk_mcp_server(name="utilities", version="1.0.0", tools=[calculate])

# Async generator (streaming input)
async def message_gen():
    yield {"type": "user", "message": {"role": "user", "content": "Calculate 5 + 3"}}

# This FAILS with ProcessTransport error
async def test():
    async for msg in query(
        prompt=message_gen(),  # ❌ Streaming input fails
        options=ClaudeAgentOptions(mcp_servers={"utilities": server})
    ):
        print(msg)

asyncio.run(test())

Workaround

Using a simple string prompt instead of async generator works:

# This WORKS
async def test():
    async for msg in query(
        prompt="Calculate 5 + 3",  # ✅ Simple string works
        options=ClaudeAgentOptions(mcp_servers={"utilities": server})
    ):
        print(msg)

asyncio.run(test())

Expected Behavior

Per the Custom Tools documentation, SDK MCP servers should work with streaming input mode (async generators).

Additional Notes

• The issue occurs with SDK MCP servers only (in-process, created via create_sdk_mcp_server())
• External MCP servers (stdio/HTTP subprocess) likely unaffected (different transport)
• The bug appears to be in subprocess_cli.py attempting ProcessTransport for in-process tools
• Workaround is functional but contradicts documentation

Related

• Documentation claims streaming required: https://docs.anthropic.com/en/api/agent-sdk/custom-tools
• Similar issue may affect TypeScript SDK (untested)

[modesty]

+1, same problem