Run the Claude Code CLI and capture token usage.
(prompt: str, workdir: str, model: str = "opus")
| 84 | |
| 85 | |
| 86 | def run_claude_cli(prompt: str, workdir: str, model: str = "opus") -> RunResult: |
| 87 | """Run the Claude Code CLI and capture token usage.""" |
| 88 | try: |
| 89 | result = subprocess.run( |
| 90 | [ |
| 91 | "claude", |
| 92 | "-p", |
| 93 | "--output-format", "json", |
| 94 | "--dangerously-skip-permissions", |
| 95 | "--model", model, |
| 96 | prompt, |
| 97 | ], |
| 98 | capture_output=True, |
| 99 | text=True, |
| 100 | cwd=workdir, |
| 101 | timeout=120, |
| 102 | ) |
| 103 | |
| 104 | if result.returncode != 0 and not result.stdout: |
| 105 | return RunResult( |
| 106 | tool="claude", |
| 107 | prompt=prompt, |
| 108 | usage=TokenUsage(0, 0, 0, 0), |
| 109 | success=False, |
| 110 | output="", |
| 111 | error=result.stderr or f"Exit code {result.returncode}", |
| 112 | ) |
| 113 | |
| 114 | # Parse JSON output |
| 115 | data = json.loads(result.stdout) |
| 116 | usage = data.get("usage", {}) |
| 117 | |
| 118 | token_usage = TokenUsage( |
| 119 | input_tokens=usage.get("input_tokens", 0), |
| 120 | output_tokens=usage.get("output_tokens", 0), |
| 121 | cache_read_tokens=usage.get("cache_read_input_tokens", 0), |
| 122 | cache_creation_tokens=usage.get("cache_creation_input_tokens", 0), |
| 123 | total_cost_usd=data.get("total_cost_usd"), |
| 124 | duration_ms=data.get("duration_ms"), |
| 125 | ) |
| 126 | |
| 127 | return RunResult( |
| 128 | tool="claude", |
| 129 | prompt=prompt, |
| 130 | usage=token_usage, |
| 131 | success=not data.get("is_error", False), |
| 132 | output=data.get("result", ""), |
| 133 | ) |
| 134 | |
| 135 | except subprocess.TimeoutExpired: |
| 136 | return RunResult( |
| 137 | tool="claude", |
| 138 | prompt=prompt, |
| 139 | usage=TokenUsage(0, 0, 0, 0), |
| 140 | success=False, |
| 141 | output="", |
| 142 | error="Timeout after 120s", |
| 143 | ) |
no test coverage detected