"""Subprocess transport implementation using Claude Code CLI.""" import atexit import json import logging import os import platform import re import shutil import signal from collections.abc import AsyncIterable, AsyncIterator from contextlib import suppress from pathlib import Path from subprocess import PIPE from typing import Any, cast import anyio from anyio.abc import Process from anyio.streams.text import TextReceiveStream, TextSendStream from ..._errors import CLIConnectionError, CLINotFoundError, ProcessError from ..._errors import CLIJSONDecodeError as SDKJSONDecodeError from ..._version import __version__ from ...types import ClaudeAgentOptions, SystemPromptFile, SystemPromptPreset from .._task_compat import TaskHandle, spawn_detached from . import Transport logger = logging.getLogger(__name__) _DEFAULT_MAX_BUFFER_SIZE = 1024 * 1024 # 1MB buffer limit MINIMUM_CLAUDE_CODE_VERSION = "2.0.0" # 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 # prevents orphaned `claude` processes from leaking when callers crash or exit # before awaiting close(). _ACTIVE_CHILDREN: set[Process] = set() def _kill_active_children() -> None: for p in list(_ACTIVE_CHILDREN): with suppress(Exception): p.send_signal(signal.SIGTERM) # On Windows anyio maps this to terminate() _ACTIVE_CHILDREN.clear() atexit.register(_kill_active_children) class SubprocessCLITransport(Transport): """Subprocess transport using Claude Code CLI.""" def __init__( self, prompt: str | AsyncIterable[dict[str, Any]], options: ClaudeAgentOptions, ): self._prompt = prompt # Always use streaming mode internally (matching TypeScript SDK) # This allows agents and other large configs to be sent via initialize request self._is_streaming = True self._options = options self._cli_path: str | None = ( str(options.cli_path) if options.cli_path is not None else None ) self._cwd = str(options.cwd) if options.cwd else None self._process: Process | None = None self._stdout_stream: TextReceiveStream | None = None self._stdin_stream: TextSendStream | None = None self._stderr_stream: TextReceiveStream | None = None self._stderr_task: TaskHandle | None = None self._ready = False self._exit_error: Exception | None = None # Track process exit errors self._max_buffer_size = ( options.max_buffer_size if options.max_buffer_size is not None else _DEFAULT_MAX_BUFFER_SIZE ) self._write_lock: anyio.Lock = anyio.Lock() def _find_cli(self) -> str: """Find Claude Code CLI binary.""" # First, check for bundled CLI bundled_cli = self._find_bundled_cli() if bundled_cli: return bundled_cli # Fall back to system-wide search if cli := shutil.which("claude"): return cli locations = [ Path.home() / ".npm-global/bin/claude", Path("/usr/local/bin/claude"), Path.home() / ".local/bin/claude", Path.home() / "node_modules/.bin/claude", Path.home() / ".yarn/bin/claude", Path.home() / ".claude/local/claude", ] for path in locations: if path.exists() and path.is_file(): return str(path) raise CLINotFoundError( "Claude Code not found. Install with:\n" " npm install -g @anthropic-ai/claude-code\n" "\nIf already installed locally, try:\n" ' export PATH="$HOME/node_modules/.bin:$PATH"\n' "\nOr provide the path via ClaudeAgentOptions:\n" " ClaudeAgentOptions(cli_path='/path/to/claude')" ) def _find_bundled_cli(self) -> str | None: """Find bundled CLI binary if it exists.""" # Determine the CLI binary name based on platform cli_name = "claude.exe" if platform.system() == "Windows" else "claude" # Get the path to the bundled CLI # The _bundled directory is in the same package as this module bundled_path = Path(__file__).parent.parent.parent / "_bundled" / cli_name if bundled_path.exists() and bundled_path.is_file(): logger.info(f"Using bundled Claude Code CLI: {bundled_path}") return str(bundled_path) return None def _build_settings_value(self) -> str | None: """Build settings value, merging sandbox settings if provided. Returns the settings value as either: - A JSON string (if sandbox is provided or settings is JSON) - A file path (if only settings path is provided without sandbox) - None if neither settings nor sandbox is provided """ has_settings = self._options.settings is not None has_sandbox = self._options.sandbox is not None if not has_settings and not has_sandbox: return None # If only settings path and no sandbox, pass through as-is if has_settings and not has_sandbox: return self._options.settings # If we have sandbox settings, we need to merge into a JSON object settings_obj: dict[str, Any] = {} if has_settings: assert self._options.settings is not None settings_str = self._options.settings.strip() # Check if settings is a JSON string or a file path if settings_str.startswith("{") and settings_str.endswith("}"): # Parse JSON string try: settings_obj = json.loads(settings_str) except json.JSONDecodeError: # If parsing fails, treat as file path logger.warning( f"Failed to parse settings as JSON, treating as file path: {settings_str}" ) # Read the file settings_path = Path(settings_str) if settings_path.exists(): with settings_path.open(encoding="utf-8") as f: settings_obj = json.load(f) else: # It's a file path - read and parse settings_path = Path(settings_str) if settings_path.exists(): with settings_path.open(encoding="utf-8") as f: settings_obj = json.load(f) else: logger.warning(f"Settings file not found: {settings_path}") # Merge sandbox settings if has_sandbox: settings_obj["sandbox"] = self._options.sandbox return json.dumps(settings_obj) def _apply_skills_defaults( self, ) -> tuple[list[str], list[str] | None]: """Compute effective allowed_tools and setting_sources for skills. When ``options.skills`` is ``"all"``, injects the bare ``Skill`` tool; when it is a list, injects ``Skill(name)`` for each entry. In either case ``setting_sources`` defaults to ``["user", "project"]`` when unset so the CLI discovers installed skills without the caller having to wire up both options manually. ``None`` is a no-op. Does not mutate the original options object. """ allowed_tools: list[str] = list(self._options.allowed_tools) setting_sources: list[str] | None = ( list(self._options.setting_sources) if self._options.setting_sources is not None else None ) skills = self._options.skills if skills is None: return allowed_tools, setting_sources if skills == "all": if "Skill" not in allowed_tools: allowed_tools.append("Skill") else: for name in skills: pattern = f"Skill({name})" if pattern not in allowed_tools: allowed_tools.append(pattern) if setting_sources is None: setting_sources = ["user", "project"] return allowed_tools, setting_sources def _build_command(self) -> list[str]: """Build CLI command with arguments.""" if self._cli_path is None: raise CLINotFoundError("CLI path not resolved. Call connect() first.") cmd = [self._cli_path, "--output-format", "stream-json", "--verbose"] if self._options.system_prompt is None: cmd.extend(["--system-prompt", ""]) elif isinstance(self._options.system_prompt, str): cmd.extend(["--system-prompt", self._options.system_prompt]) else: sp = self._options.system_prompt if sp.get("type") == "file": cmd.extend(["--system-prompt-file", cast(SystemPromptFile, sp)["path"]]) elif sp.get("type") == "preset" and "append" in sp: cmd.extend( ["--append-system-prompt", cast(SystemPromptPreset, sp)["append"]] ) # Handle tools option (base set of tools) if self._options.tools is not None: tools = self._options.tools if isinstance(tools, list): if len(tools) == 0: cmd.extend(["--tools", ""]) else: cmd.extend(["--tools", ",".join(tools)]) else: # Preset object - 'claude_code' preset maps to 'default' cmd.extend(["--tools", "default"]) effective_allowed_tools, effective_setting_sources = ( self._apply_skills_defaults() ) if effective_allowed_tools: cmd.extend(["--allowedTools", ",".join(effective_allowed_tools)]) if self._options.max_turns: cmd.extend(["--max-turns", str(self._options.max_turns)]) if self._options.max_budget_usd is not None: cmd.extend(["--max-budget-usd", str(self._options.max_budget_usd)]) if self._options.disallowed_tools: cmd.extend(["--disallowedTools", ",".join(self._options.disallowed_tools)]) if self._options.task_budget is not None: cmd.extend(["--task-budget", str(self._options.task_budget["total"])]) if self._options.model: cmd.extend(["--model", self._options.model]) if self._options.fallback_model: cmd.extend(["--fallback-model", self._options.fallback_model]) if self._options.betas: cmd.extend(["--betas", ",".join(self._options.betas)]) if self._options.permission_prompt_tool_name: cmd.extend( ["--permission-prompt-tool", self._options.permission_prompt_tool_name] ) if self._options.permission_mode: cmd.extend(["--permission-mode", self._options.permission_mode]) if self._options.continue_conversation: cmd.append("--continue") if self._options.resume: cmd.extend(["--resume", self._options.resume]) if self._options.session_id: cmd.extend(["--session-id", self._options.session_id]) # Handle settings and sandbox: merge sandbox into settings if both are provided settings_value = self._build_settings_value() if settings_value: cmd.extend(["--settings", settings_value]) if self._options.add_dirs: # Convert all paths to strings and add each directory for directory in self._options.add_dirs: cmd.extend(["--add-dir", str(directory)]) if self._options.mcp_servers: if isinstance(self._options.mcp_servers, dict): # Process all servers, stripping instance field from SDK servers servers_for_cli: dict[str, Any] = {} for name, config in self._options.mcp_servers.items(): if isinstance(config, dict) and config.get("type") == "sdk": # For SDK servers, pass everything except the instance field sdk_config: dict[str, object] = { k: v for k, v in config.items() if k != "instance" } servers_for_cli[name] = sdk_config else: # For external servers, pass as-is servers_for_cli[name] = config # Pass all servers to CLI if servers_for_cli: cmd.extend( [ "--mcp-config", json.dumps({"mcpServers": servers_for_cli}), ] ) else: # String or Path format: pass directly as file path or JSON string cmd.extend(["--mcp-config", str(self._options.mcp_servers)]) if self._options.include_partial_messages: cmd.append("--include-partial-messages") if self._options.include_hook_events: cmd.append("--include-hook-events") if self._options.strict_mcp_config: cmd.append("--strict-mcp-config") if self._options.fork_session: cmd.append("--fork-session") if self._options.session_store is not None: cmd.append("--session-mirror") # Agents are always sent via initialize request (matching TypeScript SDK) # No --agents CLI flag needed if effective_setting_sources is not None: cmd.append(f"--setting-sources={','.join(effective_setting_sources)}") # Add plugin directories if self._options.plugins: for plugin in self._options.plugins: if plugin["type"] == "local": cmd.extend(["--plugin-dir", plugin["path"]]) else: raise ValueError(f"Unsupported plugin type: {plugin['type']}") # Add extra args for future CLI flags for flag, value in self._options.extra_args.items(): if value is None: # Boolean flag without value cmd.append(f"--{flag}") else: # Flag with value cmd.extend([f"--{flag}", str(value)]) # Resolve thinking config -> --thinking / --max-thinking-tokens # `thinking` takes precedence over the deprecated `max_thinking_tokens` if self._options.thinking is not None: t = self._options.thinking if t["type"] == "adaptive": cmd.extend(["--thinking", "adaptive"]) elif t["type"] == "enabled": cmd.extend(["--max-thinking-tokens", str(t["budget_tokens"])]) elif t["type"] == "disabled": cmd.extend(["--thinking", "disabled"]) # Narrow off the Disabled variant first so mypy knows `t["display"]` is a str # rather than widening to `object` across the union. if t["type"] != "disabled" and "display" in t: cmd.extend(["--thinking-display", t["display"]]) elif self._options.max_thinking_tokens is not None: cmd.extend( ["--max-thinking-tokens", str(self._options.max_thinking_tokens)] ) if self._options.effort is not None: cmd.extend(["--effort", self._options.effort]) # Extract schema from output_format structure if provided # Expected: {"type": "json_schema", "schema": {...}} if ( self._options.output_format is not None and isinstance(self._options.output_format, dict) and self._options.output_format.get("type") == "json_schema" ): schema = self._options.output_format.get("schema") if schema is not None: cmd.extend(["--json-schema", json.dumps(schema)]) # Always use streaming mode with stdin (matching TypeScript SDK) # This allows agents and other large configs to be sent via initialize request cmd.extend(["--input-format", "stream-json"]) return cmd async def connect(self) -> None: """Start subprocess.""" if self._process: return if self._cli_path is None: self._cli_path = await anyio.to_thread.run_sync(self._find_cli) if not os.environ.get("CLAUDE_AGENT_SDK_SKIP_VERSION_CHECK"): await self._check_claude_version() cmd = self._build_command() try: # Merge environment variables. CLAUDE_CODE_ENTRYPOINT defaults to # sdk-py regardless of inherited process env; options.env can override # it. CLAUDE_AGENT_SDK_VERSION is always set by the SDK. # Filter out CLAUDECODE so SDK-spawned subprocesses don't think # they're running inside a Claude Code parent (see #573). inherited_env = {k: v for k, v in os.environ.items() if k != "CLAUDECODE"} process_env = { **inherited_env, "CLAUDE_CODE_ENTRYPOINT": "sdk-py", **self._options.env, "CLAUDE_AGENT_SDK_VERSION": __version__, } # Propagate active OTEL trace context to the CLI so its spans # parent under the caller's distributed trace. No-op if # opentelemetry-api is not installed or there's no active span. try: from opentelemetry import propagate carrier: dict[str, str] = {} propagate.inject(carrier) if "traceparent" in carrier: # Active span present: scrub stale inherited W3C context # (CI/k8s ambient env) before writing the fresh values, so # an inherited TRACESTATE isn't paired with a new # TRACEPARENT. Explicit ClaudeAgentOptions.env always wins. # Gate on the traceparent key (not carrier truthiness) so a # baggage-only / non-W3C carrier doesn't scrub a valid # inherited TRACEPARENT. for key in ("TRACEPARENT", "TRACESTATE"): if key not in self._options.env: process_env.pop(key, None) for k, v in carrier.items(): key = k.upper() if key not in self._options.env: process_env[key] = v except Exception: # noqa: BLE001 - best-effort tracing must never break connect() logger.debug("OTEL trace context injection failed", exc_info=True) # Enable file checkpointing if requested if self._options.enable_file_checkpointing: process_env["CLAUDE_CODE_ENABLE_SDK_FILE_CHECKPOINTING"] = "true" if self._cwd: process_env["PWD"] = self._cwd # Pipe stderr only when the caller registered a callback. stderr_dest = PIPE if self._options.stderr is not None else None self._process = await anyio.open_process( cmd, stdin=PIPE, stdout=PIPE, stderr=stderr_dest, cwd=self._cwd, env=process_env, user=self._options.user, ) _ACTIVE_CHILDREN.add(self._process) if self._process.stdout: self._stdout_stream = TextReceiveStream(self._process.stdout) # Setup stderr stream if piped if stderr_dest is PIPE and self._process.stderr: self._stderr_stream = TextReceiveStream(self._process.stderr) # Spawn the stderr reader via spawn_detached (not a manually- # entered TaskGroup) so cleanup has no trio task-affinity — # same pattern as Query._read_task. self._stderr_task = spawn_detached(self._handle_stderr()) # Setup stdin for streaming (always used now) if self._process.stdin: self._stdin_stream = TextSendStream(self._process.stdin) self._ready = True except FileNotFoundError as e: # Check if the error comes from the working directory or the CLI if self._cwd and not Path(self._cwd).exists(): error = CLIConnectionError( f"Working directory does not exist: {self._cwd}" ) self._exit_error = error raise error from e error = CLINotFoundError(f"Claude Code not found at: {self._cli_path}") self._exit_error = error raise error from e except Exception as e: error = CLIConnectionError(f"Failed to start Claude Code: {e}") self._exit_error = error raise error from e async def _handle_stderr(self) -> None: """Handle stderr stream - read and invoke callbacks.""" if not self._stderr_stream: return try: async for line in self._stderr_stream: line_str = line.rstrip() if not line_str: continue # Call the stderr callback if provided if self._options.stderr: self._options.stderr(line_str) except anyio.ClosedResourceError: pass # Stream closed, exit normally except Exception: pass # Ignore other errors during stderr reading async def close(self) -> None: """Close the transport and clean up resources.""" if not self._process: self._ready = False return # Cancel stderr reader if active if self._stderr_task is not None and not self._stderr_task.done(): self._stderr_task.cancel() with suppress(Exception): await self._stderr_task.wait() self._stderr_task = None # Close stdin stream (acquire lock to prevent race with concurrent writes) async with self._write_lock: self._ready = False # Set inside lock to prevent TOCTOU with write() if self._stdin_stream: with suppress(Exception): await self._stdin_stream.aclose() self._stdin_stream = None if self._stderr_stream: with suppress(Exception): await self._stderr_stream.aclose() self._stderr_stream = None # Wait for graceful shutdown after stdin EOF, then terminate if needed. # The subprocess needs time to flush its session file after receiving # EOF on stdin. Without this grace period, SIGTERM can interrupt the # write and cause the last assistant message to be lost (see #625). try: if self._process.returncode is None: try: with anyio.fail_after(5): await self._process.wait() except TimeoutError: # Graceful shutdown timed out — force terminate with suppress(ProcessLookupError): self._process.terminate() try: with anyio.fail_after(5): await self._process.wait() except TimeoutError: # SIGTERM handler blocked — force kill (SIGKILL) with suppress(ProcessLookupError): self._process.kill() with suppress(Exception): await self._process.wait() finally: _ACTIVE_CHILDREN.discard(self._process) self._process = None self._stdout_stream = None self._stdin_stream = None self._stderr_stream = None self._exit_error = None async def write(self, data: str) -> None: """Write raw data to the transport.""" async with self._write_lock: # All checks inside lock to prevent TOCTOU races with close()/end_input() if not self._ready or not self._stdin_stream: raise CLIConnectionError("ProcessTransport is not ready for writing") if self._process and self._process.returncode is not None: raise CLIConnectionError( f"Cannot write to terminated process (exit code: {self._process.returncode})" ) if self._exit_error: raise CLIConnectionError( f"Cannot write to process that exited with error: {self._exit_error}" ) from self._exit_error try: await self._stdin_stream.send(data) except Exception as e: self._ready = False self._exit_error = CLIConnectionError( f"Failed to write to process stdin: {e}" ) raise self._exit_error from e async def end_input(self) -> None: """End the input stream (close stdin).""" async with self._write_lock: if self._stdin_stream: with suppress(Exception): await self._stdin_stream.aclose() self._stdin_stream = None def read_messages(self) -> AsyncIterator[dict[str, Any]]: """Read and parse messages from the transport.""" return self._read_messages_impl() async def _read_messages_impl(self) -> AsyncIterator[dict[str, Any]]: """Internal implementation of read_messages.""" if not self._process or not self._stdout_stream: raise CLIConnectionError("Not connected") json_buffer = "" # Process stdout messages try: async for line in self._stdout_stream: line_str = line.strip() if not line_str: continue # Accumulate partial JSON until we can parse it # Note: TextReceiveStream can truncate long lines, so we need to buffer # and speculatively parse until we get a complete JSON object json_lines = line_str.split("\n") for json_line in json_lines: json_line = json_line.strip() if not json_line: continue # Skip non-JSON lines (e.g. [SandboxDebug]) when not # mid-parse — they corrupt the buffer otherwise (#347). if not json_buffer and not json_line.startswith("{"): logger.debug( "Skipping non-JSON line from CLI stdout: %s", json_line[:200], ) continue # Keep accumulating partial JSON until we can parse it json_buffer += json_line if len(json_buffer) > self._max_buffer_size: buffer_length = len(json_buffer) json_buffer = "" raise SDKJSONDecodeError( f"JSON message exceeded maximum buffer size of {self._max_buffer_size} bytes", ValueError( f"Buffer size {buffer_length} exceeds limit {self._max_buffer_size}" ), ) try: data = json.loads(json_buffer) json_buffer = "" yield data except json.JSONDecodeError: # We are speculatively decoding the buffer until we get # a full JSON object. If there is an actual issue, we # raise an error after exceeding the configured limit. continue except anyio.ClosedResourceError: pass except GeneratorExit: # Client disconnected pass # Check process completion and handle errors try: returncode = await self._process.wait() except Exception: returncode = -1 # Use exit code for error detection if returncode is not None and returncode != 0: self._exit_error = ProcessError( f"Command failed with exit code {returncode}", exit_code=returncode, stderr="Check stderr output for details", ) raise self._exit_error async def _check_claude_version(self) -> None: """Check Claude Code version and warn if below minimum.""" if self._cli_path is None: raise CLINotFoundError("CLI path not resolved. Call connect() first.") version_process = None try: with anyio.fail_after(2): # 2 second timeout version_process = await anyio.open_process( [self._cli_path, "-v"], stdout=PIPE, stderr=PIPE, ) if version_process.stdout: stdout_bytes = await version_process.stdout.receive() version_output = stdout_bytes.decode().strip() match = re.match(r"([0-9]+\.[0-9]+\.[0-9]+)", version_output) if match: version = match.group(1) version_parts = [int(x) for x in version.split(".")] min_parts = [ int(x) for x in MINIMUM_CLAUDE_CODE_VERSION.split(".") ] if version_parts < min_parts: logger.warning( "Claude Code version %s at %s is unsupported in the Agent SDK. " "Minimum required version is %s. " "Some features may not work correctly.", version, self._cli_path, MINIMUM_CLAUDE_CODE_VERSION, ) except Exception: pass finally: if version_process: with suppress(Exception): version_process.terminate() with suppress(Exception): await version_process.wait() def is_ready(self) -> bool: """Check if transport is ready for communication.""" return self._ready