"""Claude Code CLI client — invokes `claude -p` as a subprocess.""" from __future__ import annotations import json import logging import shutil import subprocess from src.llm.client import LLMClient, load_config logger = logging.getLogger(__name__) class ClaudeCLIClient(LLMClient): """LLM client that shells out to the Claude Code CLI. Uses `claude -p` (print mode) with JSON output. Expert knowledge is injected via --system-prompt so it benefits from prompt caching across calls that share the same knowledge file. """ def __init__(self, config: dict | None = None): self.config = config or load_config() llm_cfg = self.config["llm"] self.model = llm_cfg.get("model", "sonnet") self.max_tokens = llm_cfg.get("max_tokens", 4096) self.claude_bin = llm_cfg.get("claude_bin", "claude") resolved = shutil.which(self.claude_bin) if resolved is None: raise FileNotFoundError( f"Claude Code CLI not found at '{self.claude_bin}'. " "Install it or set llm.claude_bin in config.yaml." ) self.claude_bin = resolved def call( self, prompt_template: str, variables: dict, expert_knowledge: str = "", seed: int | None = None, ) -> str: if seed is not None: raise NotImplementedError( "Anthropic API does not yet expose a deterministic seed parameter. " "Issue A reserves the seed kwarg for the local backend only." ) variables_with_knowledge = {**variables, "expert_knowledge": expert_knowledge} prompt = prompt_template.format(**variables_with_knowledge) cmd = [ self.claude_bin, "-p", "--output-format", "json", "--model", self.model, "--bare", "--no-session-persistence", "--max-tokens", str(self.max_tokens), ] if expert_knowledge: cmd.extend(["--system-prompt", expert_knowledge]) logger.debug(f"Claude CLI call: model={self.model}, prompt_len={len(prompt)}") result = subprocess.run( cmd, input=prompt, capture_output=True, text=True, timeout=300, ) if result.returncode != 0: logger.error(f"Claude CLI failed (rc={result.returncode}): {result.stderr[:500]}") raise RuntimeError(f"Claude CLI exited with code {result.returncode}: {result.stderr[:300]}") return self._parse_output(result.stdout) def _parse_output(self, stdout: str) -> str: """Extract the response text from Claude CLI JSON output.""" try: data = json.loads(stdout) except json.JSONDecodeError: # If not JSON, treat the raw stdout as the response (text mode fallback) return stdout.strip() if data.get("is_error"): error_msg = data.get("result", "Unknown CLI error") raise RuntimeError(f"Claude CLI error: {error_msg}") return data.get("result", "")