ProcessTransport Error When Using MCP Servers

[abrichr]

Description

When MCP servers are provided to the Claude Code SDK, the subprocess transport fails to initialize with a "ProcessTransport is not ready for writing" error. The same code works perfectly without MCP servers. This issue occurs on both macOS and Linux (Docker container) environments.

Environment

Tested and reproduced on:

macOS (Host)

• OS: macOS
• Python Version: 3.12
• claude-code-sdk version: 0.0.23
• @anthropic-ai/claude-code CLI version: 1.0.119

Linux (Docker container)

• OS: Linux (Docker container)
• Python Version: 3.10
• claude-code-sdk version: 0.0.23
• @anthropic-ai/claude-code CLI version: 1.0.119

Reproduction Steps

1. Run the provided script in a Docker container
2. Observe that the test without MCP servers passes
3. Observe that the test with MCP servers fails with ProcessTransport error

Minimal Reproduction Code

#!/usr/bin/env python
"""
Minimal reproducible example for Claude Code SDK MCP server bug.

Issue: When MCP servers are provided to the SDK, the subprocess transport
fails with "ProcessTransport is not ready for writing" error.

Environment:
- Python 3.10
- claude-code-sdk==0.0.23
- @anthropic-ai/claude-code@1.0.119
- Running in Docker container (Linux)

Error occurs when:
1. MCP servers are provided in ClaudeCodeOptions
2. Using streaming input (required for MCP)
3. The subprocess CLI transport tries to initialize

The same code works fine without MCP servers.
"""

import asyncio
import os
from claude_code_sdk import (
    ClaudeCodeOptions,
    query,
    tool,
    create_sdk_mcp_server
)


# Create a minimal MCP tool for testing
@tool(
    name="test_tool",
    description="A simple test tool",
    input_schema={
        "type": "object",
        "properties": {
            "message": {"type": "string"}
        },
        "required": ["message"]
    }
)
async def test_tool(args: dict) -> dict:
    """Simple test tool that echoes a message."""
    return {
        "content": [
            {
                "type": "text",
                "text": f"Test response: {args.get('message', 'no message')}"
            }
        ]
    }


async def test_without_mcp():
    """Test WITHOUT MCP servers - this works fine."""
    print("\n=== Test WITHOUT MCP servers ===")
    
    options = ClaudeCodeOptions(
        cwd="/tmp",  # Use temp directory
        max_turns=1,
        permission_mode="acceptEdits",
        allowed_tools=["Read", "Write"],
    )
    
    try:
        message_count = 0
        async for message in query(
            prompt="Say hello",
            options=options
        ):
            message_count += 1
            print(f"  Message {message_count}: {type(message).__name__}")
            if message_count >= 3:
                break
        print(f"✅ SUCCESS: Got {message_count} messages without MCP\n")
        return True
    except Exception as e:
        print(f"❌ FAILED: {type(e).__name__}: {e}\n")
        return False


async def test_with_mcp():
    """Test WITH MCP servers - this fails with ProcessTransport error."""
    print("=== Test WITH MCP servers ===")
    
    # Create MCP server
    mcp_server = create_sdk_mcp_server(
        name="test-server",
        version="1.0.0",
        tools=[test_tool]
    )
    
    options = ClaudeCodeOptions(
        cwd="/tmp",  # Use temp directory
        max_turns=1,
        permission_mode="acceptEdits",
        allowed_tools=[
            "Read",
            "Write",
            "mcp__test-server__test_tool",  # MCP tool in correct format
        ],
        mcp_servers={
            "test-server": mcp_server
        }
    )
    
    # Create streaming input (required for MCP servers)
    async def stream_input():
        yield {
            "type": "user",
            "message": {
                "role": "user",
                "content": "Say hello and list available tools"
            }
        }
    
    try:
        message_count = 0
        async for message in query(
            prompt=stream_input(),  # Streaming input for MCP
            options=options
        ):
            message_count += 1
            print(f"  Message {message_count}: {type(message).__name__}")
            if message_count >= 3:
                break
        print(f"✅ SUCCESS: Got {message_count} messages with MCP\n")
        return True
    except Exception as e:
        print(f"❌ FAILED: {type(e).__name__}: {e}")
        
        # Print the full error for debugging
        if "TaskGroup" in str(type(e)):
            print("\nFull error details:")
            import traceback
            traceback.print_exc()
        print()
        return False


async def main():
    """Run both tests to demonstrate the issue."""
    print("Claude Code SDK MCP Server Bug Reproduction")
    print("=" * 50)
    
    # Set API key from environment
    api_key = os.environ.get("ANTHROPIC_API_KEY")
    if not api_key:
        print("ERROR: Please set ANTHROPIC_API_KEY environment variable")
        return
    
    # Test without MCP (should work)
    without_mcp_success = await test_without_mcp()
    
    # Test with MCP (will fail with ProcessTransport error)
    with_mcp_success = await test_with_mcp()
    
    # Summary
    print("=" * 50)
    print("SUMMARY:")
    print(f"  Without MCP servers: {'✅ PASSED' if without_mcp_success else '❌ FAILED'}")
    print(f"  With MCP servers:    {'✅ PASSED' if with_mcp_success else '❌ FAILED'}")
    
    if not with_mcp_success:
        print("\nEXPECTED ERROR:")
        print("  The MCP test fails with: ProcessTransport is not ready for writing")
        print("  This occurs when the SDK tries to initialize the subprocess CLI")
        print("  transport with MCP servers in a containerized environment.")


if __name__ == "__main__":
    # Run the reproduction script
    asyncio.run(main())

Expected Behavior

The SDK should successfully initialize the subprocess transport and execute the query with MCP servers enabled.

Actual Behavior

The subprocess transport fails during initialization with the following error:

asyncio.exceptions.CancelledError: Cancelled by cancel scope [...]

During handling of the above exception, another exception occurred:

claude_code_sdk._errors.CLIConnectionError: ProcessTransport is not ready for writing

Full Error Traceback

Traceback (most recent call last):
  File "/app/.venv/lib/python3.10/site-packages/claude_code_sdk/_internal/query.py", line 271, in _handle_control_request
    await self.transport.write(json.dumps(success_response) + "\n")
  File "/app/.venv/lib/python3.10/site-packages/claude_code_sdk/_internal/transport/subprocess_cli.py", line 273, in write
    raise CLIConnectionError("ProcessTransport is not ready for writing")
claude_code_sdk._errors.CLIConnectionError: ProcessTransport is not ready for writing

Additional Context

• This issue only occurs when MCP servers are provided in the ClaudeCodeOptions
• The same code works perfectly without MCP servers
• Occurs on both macOS host and Linux containers
• The issue appears to be related to subprocess initialization timing when MCP servers are configured
• Using streaming input (which is required for MCP servers) vs simple string input doesn't affect the outcome
• The subprocess creation gets cancelled during the _make_subprocess_transport phase

Related Issues

• The In Process MCP is not working at all #147
• [BUG] In-process MCP servers bug in Claude Code TypeScript SDK claude-code#7279

[copumpkin]

Yeah, I've been trying to get in-process MCP servers to work too and even the basic calculator examples in the documentation immediately appear in the failed state.

[sk8erwitskil]

I am seeing the exact same issue. has anyone figured out a workaround or fix?

[abrichr]

The workaround is to use traditional external MCP servers that Claude CLI spawns as subprocesses. This avoids the SDK's ProcessTransport initialization entirely.

Untested:

mcp_tool.py:

# mcp_tool.py
import asyncio
import json
import sys

async def main():
    while True:
        line = await asyncio.get_event_loop().run_in_executor(None, sys.stdin.readline)
        if not line:
            break
        
        request = json.loads(line)
        method = request.get("method")
        
        if method == "initialize":
            result = {
                "protocolVersion": "2024-11-05",
                "serverInfo": {"name": "my-server", "version": "1.0.0"},
                "capabilities": {"tools": {}}
            }
        elif method == "tools/list":
            result = {
                "tools": [{
                    "name": "my_tool",
                    "description": "Does something",
                    "inputSchema": {"type": "object", "properties": {}}
                }]
            }
        elif method == "tools/call":
            # Implement tool logic
            result = {"content": [{"type": "text", "text": "Result"}]}
        
        response = {"jsonrpc": "2.0", "id": request.get("id"), "result": result}
        print(json.dumps(response), flush=True)

if __name__ == "__main__":
    asyncio.run(main())

.mcp.json:

{
  "mcpServers": {
    "my-server": {
      "command": "python",
      "args": ["-m", "mcp_tool"]
    }
  }
}

Pass config to Claude CLI:

cmd = [
    "claude",
    "--input-format", "stream-json",
    "--output-format", "stream-json",
    "--mcp-config", "/path/to/.mcp.json",
    "--allowedTools", "Read,Write,mcp__my-server__my_tool"
]

The tool is accessed as mcp__<server-name>__<tool-name> in the allowedTools list.

[dltn]

Hi, to use custom tools, you need to use ClaudeSDKClient and not query. See: https://docs.claude.com/en/api/agent-sdk/python#choosing-between-query-and-claudesdkclient