CLIConnectionError when using SDK MCP servers with string prompts in query()

Summary

query() crashes with ExceptionGroup: CLIConnectionError: ProcessTransport is not ready for writing when called with a string prompt and SDK MCP servers (via create_sdk_mcp_server()). The AsyncIterable prompt path works correctly. Crash is 100% reproducible.

Root Cause

Race condition in _internal/client.py (lines 122-134). The string-prompt code path writes the user message and immediately closes stdin via transport.end_input(). However, the bundled CLI is still sending MCP initialization control_request messages (initialize, notifications/initialized, tools/list for each SDK server). When the SDK tries to respond to these via transport.write(), stdin is already closed.

The AsyncIterable code path in stream_input() (_internal/query.py, lines 585-597) correctly handles this — it checks for self.sdk_mcp_servers and waits for _first_result_event before calling end_input(). The string-prompt path skips this protection entirely.

Affected code (_internal/client.py:122-134):

if isinstance(prompt, str):
    user_message = {
        "type": "user",
        "session_id": "",
        "message": {"role": "user", "content": prompt},
        "parent_tool_use_id": None,
    }
    await chosen_transport.write(json.dumps(user_message) + "\n")
    await chosen_transport.end_input()  # <-- closes stdin while MCP init is still running

Protected code (_internal/query.py:585-597):

if self.sdk_mcp_servers or has_hooks:
    with anyio.move_on_after(self._stream_close_timeout):
        await self._first_result_event.wait()
await self.transport.end_input()  # <-- waits for MCP init to complete

Reproduction

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

@tool('test_tool', 'A test tool', {})
async def test_tool(args):
    return {'content': [{'type': 'text', 'text': 'OK'}]}

server = create_sdk_mcp_server('test', tools=[test_tool])

async def main():
    opts = ClaudeAgentOptions(
        system_prompt='Reply: OK',
        max_turns=1,
        mcp_servers={'test': server},
    )
    # This crashes 100% of the time:
    async for msg in query(prompt='Hello', options=opts):
        if isinstance(msg, ResultMessage):
            print(msg.result)

asyncio.run(main())

Error:

ExceptionGroup: unhandled errors in a TaskGroup (1 sub-exception)
  sub: CLIConnectionError: ProcessTransport is not ready for writing

Workaround

Convert the string prompt to an AsyncIterable to use the safe code path:

from typing import AsyncIterator

async def prompt_as_stream(text: str) -> AsyncIterator[dict]:
    yield {
        "type": "user",
        "session_id": "",
        "message": {"role": "user", "content": text},
        "parent_tool_use_id": None,
    }

# This works:
async for msg in query(prompt=prompt_as_stream('Hello'), options=opts):
    ...

Suggested Fix

The string-prompt path in client.py should replicate the wait-for-first-result logic from stream_input() when SDK MCP servers or hooks are present.

Environment

• claude-agent-sdk 0.1.33
• Bundled CLI 2.1.37
• Python 3.14
• macOS (Darwin 24.6.0)
• More SDK MCP servers = more reliable crash (more initialization round-trips)

Impact

Any use of query() with a string prompt + SDK MCP servers crashes 100% of the time. This is the primary documented usage pattern for SDK MCP servers.