Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
76 changes: 75 additions & 1 deletion src/claude_agent_sdk/_internal/transport/subprocess_cli.py
Original file line number Diff line number Diff line change
Expand Up @@ -8,6 +8,7 @@
import re
import shutil
import signal
import tempfile
from collections.abc import AsyncIterable, AsyncIterator
from contextlib import suppress
from pathlib import Path
Expand All @@ -29,6 +30,7 @@

_DEFAULT_MAX_BUFFER_SIZE = 1024 * 1024 # 1MB buffer limit
MINIMUM_CLAUDE_CODE_VERSION = "2.0.0"
_WINDOWS_CMD_LENGTH_LIMIT = 32767

# Track live CLI subprocesses so we can terminate them when the parent Python
# process exits. This mirrors the TypeScript SDK's parent-exit cleanup and
Expand Down Expand Up @@ -77,6 +79,7 @@ def __init__(
else _DEFAULT_MAX_BUFFER_SIZE
)
self._write_lock: anyio.Lock = anyio.Lock()
self._tempfiles_to_cleanup: list[str] = []

def _find_cli(self) -> str:
"""Find Claude Code CLI binary."""
Expand Down Expand Up @@ -409,6 +412,63 @@ def _build_command(self) -> list[str]:

return cmd

def _prepare_command_for_spawn(self, cmd: list[str]) -> list[str]:
"""Rewrite oversized inline prompt args before spawning the CLI."""
if platform.system() != "Windows":
return cmd

if len(" ".join(cmd)) <= _WINDOWS_CMD_LENGTH_LIMIT:
return cmd

prepared = list(cmd)
for inline_flag, file_flag in (
("--system-prompt", "--system-prompt-file"),
("--append-system-prompt", "--append-system-prompt-file"),
):
if inline_flag not in prepared:
continue

idx = prepared.index(inline_flag)
prompt = prepared[idx + 1]
with tempfile.NamedTemporaryFile(
mode="w",
suffix=".txt",
encoding="utf-8",
delete=False,
) as prompt_file:
prompt_file.write(prompt)
prompt_file_path = prompt_file.name

self._tempfiles_to_cleanup.append(prompt_file_path)
prepared[idx] = file_flag
prepared[idx + 1] = prompt_file_path

if len(" ".join(prepared)) <= _WINDOWS_CMD_LENGTH_LIMIT:
break

return prepared

def _cleanup_tempfiles(self) -> None:
"""Remove temporary prompt files created for CLI arguments."""
for path in self._tempfiles_to_cleanup:
with suppress(FileNotFoundError):
Path(path).unlink()
self._tempfiles_to_cleanup.clear()

def _command_too_long_error(
self, cmd: list[str], cause: FileNotFoundError
) -> CLIConnectionError | None:
"""Return a clearer error for Windows command-line overflow."""
if getattr(cause, "winerror", None) != 206:
return None

cmd_length = len(" ".join(cmd))
return CLIConnectionError(
"Failed to start Claude Code: command line exceeded the Windows "
f"limit ({cmd_length} characters, limit {_WINDOWS_CMD_LENGTH_LIMIT}). "
"Use a file-backed system_prompt or reduce inline CLI arguments."
)

async def connect(self) -> None:
"""Start subprocess."""
if self._process:
Expand All @@ -420,7 +480,7 @@ async def connect(self) -> None:
if not os.environ.get("CLAUDE_AGENT_SDK_SKIP_VERSION_CHECK"):
await self._check_claude_version()

cmd = self._build_command()
cmd = self._prepare_command_for_spawn(self._build_command())
try:
# Merge environment variables. CLAUDE_CODE_ENTRYPOINT defaults to
# sdk-py regardless of inherited process env; options.env can override
Expand Down Expand Up @@ -506,13 +566,25 @@ async def connect(self) -> None:
f"Working directory does not exist: {self._cwd}"
)
self._exit_error = error
self._cleanup_tempfiles()
raise error from e
if error := self._command_too_long_error(cmd, e):
self._exit_error = error
self._cleanup_tempfiles()
raise error from e
if self._cli_path and Path(self._cli_path).is_file():
error = CLIConnectionError(f"Failed to start Claude Code: {e}")
self._exit_error = error
self._cleanup_tempfiles()
raise error from e
error = CLINotFoundError(f"Claude Code not found at: {self._cli_path}")
self._exit_error = error
self._cleanup_tempfiles()
raise error from e
except Exception as e:
error = CLIConnectionError(f"Failed to start Claude Code: {e}")
self._exit_error = error
self._cleanup_tempfiles()
raise error from e

async def _handle_stderr(self) -> None:
Expand Down Expand Up @@ -546,6 +618,7 @@ async def close(self) -> None:
"""Close the transport and clean up resources."""
if not self._process:
self._ready = False
self._cleanup_tempfiles()
return

# Cancel stderr reader if active
Expand Down Expand Up @@ -598,6 +671,7 @@ async def close(self) -> None:
self._stdin_stream = None
self._stderr_stream = None
self._exit_error = None
self._cleanup_tempfiles()

async def write(self, data: str) -> None:
"""Write raw data to the transport."""
Expand Down
98 changes: 98 additions & 0 deletions tests/test_transport.py
Original file line number Diff line number Diff line change
Expand Up @@ -4,6 +4,7 @@
import uuid
from collections.abc import AsyncIterator
from contextlib import nullcontext
from pathlib import Path
from unittest.mock import AsyncMock, MagicMock, patch

import anyio
Expand Down Expand Up @@ -183,6 +184,70 @@ def test_build_command_with_system_prompt_file(self):
assert "--system-prompt-file" in cmd
assert "/path/to/prompt.md" in cmd

def test_prepare_command_spills_long_system_prompt_on_windows(self):
"""Long inline system prompts use a temp file on Windows."""
transport = SubprocessCLITransport(
prompt="test",
options=make_options(system_prompt="x" * 40000),
)

with patch(
"claude_agent_sdk._internal.transport.subprocess_cli.platform.system",
return_value="Windows",
):
cmd = transport._prepare_command_for_spawn(transport._build_command())

assert "--system-prompt" not in cmd
assert "--system-prompt-file" in cmd
prompt_path = Path(cmd[cmd.index("--system-prompt-file") + 1])
assert prompt_path.read_text(encoding="utf-8") == "x" * 40000

anyio.run(transport.close)
assert not prompt_path.exists()

def test_prepare_command_spills_long_append_prompt_on_windows(self):
"""Long appended system prompts use a temp file on Windows."""
transport = SubprocessCLITransport(
prompt="test",
options=make_options(
system_prompt={
"type": "preset",
"preset": "claude_code",
"append": "x" * 40000,
},
),
)

with patch(
"claude_agent_sdk._internal.transport.subprocess_cli.platform.system",
return_value="Windows",
):
cmd = transport._prepare_command_for_spawn(transport._build_command())

assert "--append-system-prompt" not in cmd
assert "--append-system-prompt-file" in cmd
prompt_path = Path(cmd[cmd.index("--append-system-prompt-file") + 1])
assert prompt_path.read_text(encoding="utf-8") == "x" * 40000

anyio.run(transport.close)
assert not prompt_path.exists()

def test_prepare_command_keeps_long_system_prompt_on_non_windows(self):
"""Non-Windows platforms keep inline system prompts."""
transport = SubprocessCLITransport(
prompt="test",
options=make_options(system_prompt="x" * 40000),
)

with patch(
"claude_agent_sdk._internal.transport.subprocess_cli.platform.system",
return_value="Linux",
):
cmd = transport._prepare_command_for_spawn(transport._build_command())

assert "--system-prompt" in cmd
assert "--system-prompt-file" not in cmd

def test_build_command_with_options(self):
"""Test building CLI command with options."""
transport = SubprocessCLITransport(
Expand Down Expand Up @@ -463,6 +528,39 @@ async def _test():

anyio.run(_test)

def test_connect_winerror_206_reports_command_too_long(self, tmp_path):
"""Windows command-line overflow is not reported as missing CLI."""

async def _test():
from claude_agent_sdk._errors import CLIConnectionError, CLINotFoundError

cli_path = tmp_path / "claude.exe"
cli_path.write_text("", encoding="utf-8")
error = FileNotFoundError("The filename or extension is too long")
error.winerror = 206

with (
patch.dict(
os.environ,
{"CLAUDE_AGENT_SDK_SKIP_VERSION_CHECK": "1"},
),
patch("anyio.open_process", side_effect=error),
pytest.raises(CLIConnectionError) as exc_info,
):
transport = SubprocessCLITransport(
prompt="test",
options=make_options(
cli_path=str(cli_path),
system_prompt="x" * 40000,
),
)
await transport.connect()

assert not isinstance(exc_info.value, CLINotFoundError)
assert "command line exceeded the Windows limit" in str(exc_info.value)

anyio.run(_test)

def test_read_messages(self):
"""Test reading messages from CLI output."""
# This test is simplified to just test the transport creation
Expand Down