diff --git a/scripts/run_claude_cli_json.py b/scripts/run_claude_cli_json.py new file mode 100644 index 000000000..5ed747bec --- /dev/null +++ b/scripts/run_claude_cli_json.py @@ -0,0 +1,171 @@ +#!/usr/bin/env python3 +"""Run Claude Code CLI in print mode and emit normalized JSON.""" + +from __future__ import annotations + +import argparse +import json +import os +import shutil +import subprocess +import sys +from pathlib import Path +from typing import Any + + +DEFAULT_MAX_TURNS = 10 +DEFAULT_TIMEOUT_SECONDS = 300 +DEFAULT_MODEL = "opus" + + +def _normalize_allowed_tools(value: str | None) -> str | None: + if value is None: + return None + cleaned = value.strip() + return cleaned or None + + +def build_command( + *, + prompt: str, + model: str = DEFAULT_MODEL, + max_turns: int = DEFAULT_MAX_TURNS, + allowed_tools: str | None = None, + append_system_prompt: str | None = None, +) -> list[str]: + command = [ + "claude", + "-p", + prompt, + "--output-format", + "json", + "--max-turns", + str(max_turns), + ] + if model: + command.extend(["--model", model]) + allowed_tools = _normalize_allowed_tools(allowed_tools) + if allowed_tools: + command.extend(["--allowedTools", allowed_tools]) + if append_system_prompt: + command.extend(["--append-system-prompt", append_system_prompt]) + return command + + +def run_claude_cli( + *, + prompt: str, + workdir: str | None = None, + model: str = DEFAULT_MODEL, + max_turns: int = DEFAULT_MAX_TURNS, + allowed_tools: str | None = None, + append_system_prompt: str | None = None, + timeout_seconds: int = DEFAULT_TIMEOUT_SECONDS, +) -> dict[str, Any]: + claude_path = shutil.which("claude") + if not claude_path: + return { + "success": False, + "error": "claude CLI not found on PATH", + "command": None, + "cwd": workdir or os.getcwd(), + } + + if not prompt or not prompt.strip(): + return { + "success": False, + "error": "prompt is required", + "command": None, + "cwd": workdir or os.getcwd(), + } + + cwd = str(Path(workdir).expanduser()) if workdir else os.getcwd() + command = build_command( + prompt=prompt, + model=model, + max_turns=max_turns, + allowed_tools=allowed_tools, + append_system_prompt=append_system_prompt, + ) + + try: + completed = subprocess.run( + command, + cwd=cwd, + capture_output=True, + text=True, + timeout=timeout_seconds, + ) + except subprocess.TimeoutExpired as exc: + return { + "success": False, + "error": f"claude CLI timed out after {timeout_seconds}s", + "command": command, + "cwd": cwd, + "stdout": exc.stdout or "", + "stderr": exc.stderr or "", + } + + stdout = (completed.stdout or "").strip() + stderr = (completed.stderr or "").strip() + + parsed: dict[str, Any] | None = None + if stdout: + try: + parsed = json.loads(stdout) + except json.JSONDecodeError: + parsed = None + + success = completed.returncode == 0 and isinstance(parsed, dict) and not parsed.get("is_error", False) + result: dict[str, Any] = { + "success": success, + "command": command, + "cwd": cwd, + "claude_path": claude_path, + "exit_code": completed.returncode, + "stdout": stdout, + "stderr": stderr, + } + if parsed is not None: + result["claude_json"] = parsed + result["result"] = parsed.get("result") + result["session_id"] = parsed.get("session_id") + result["subtype"] = parsed.get("subtype") + result["total_cost_usd"] = parsed.get("total_cost_usd") + result["usage"] = parsed.get("usage") + elif stdout: + result["error"] = "claude CLI output was not valid JSON" + elif not stderr: + result["error"] = "claude CLI produced no output" + + if not success and "error" not in result: + result["error"] = stderr or result.get("result") or "claude CLI invocation failed" + return result + + +def main() -> int: + parser = argparse.ArgumentParser(description="Run Claude Code CLI in print mode and emit normalized JSON") + parser.add_argument("prompt", help="Prompt to send to Claude Code CLI") + parser.add_argument("--workdir", default=None, help="Working directory for claude CLI") + parser.add_argument("--model", default=DEFAULT_MODEL, help="Claude model alias/name (default: opus)") + parser.add_argument("--max-turns", type=int, default=DEFAULT_MAX_TURNS, help="Max turns for claude -p") + parser.add_argument("--allowed-tools", default=None, help="Pass-through value for --allowedTools") + parser.add_argument("--append-system-prompt", default=None, help="Additional system prompt text") + parser.add_argument("--timeout", type=int, default=DEFAULT_TIMEOUT_SECONDS, help="Subprocess timeout seconds") + args = parser.parse_args() + + result = run_claude_cli( + prompt=args.prompt, + workdir=args.workdir, + model=args.model, + max_turns=args.max_turns, + allowed_tools=args.allowed_tools, + append_system_prompt=args.append_system_prompt, + timeout_seconds=args.timeout, + ) + print(json.dumps(result, ensure_ascii=False)) + return 0 if result.get("success") else 1 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/tests/tools/test_claude_cli_tool.py b/tests/tools/test_claude_cli_tool.py new file mode 100644 index 000000000..690d7b73a --- /dev/null +++ b/tests/tools/test_claude_cli_tool.py @@ -0,0 +1,52 @@ +import json +import subprocess + +import tools.claude_cli_tool as claude_cli_tool + + +def test_coerce_allowed_tools_from_list(): + assert claude_cli_tool._coerce_allowed_tools(["Read", "Bash(git *)"]) == "Read,Bash(git *)" + + +def test_claude_cli_run_tool_parses_success(monkeypatch): + monkeypatch.setattr(claude_cli_tool, "BRIDGE_SCRIPT_PATH", claude_cli_tool.Path("/tmp/bridge.py")) + monkeypatch.setattr(claude_cli_tool, "_path_exists", lambda _path: True) + monkeypatch.setattr(claude_cli_tool.shutil, "which", lambda name: "/usr/bin/claude" if name == "claude" else None) + monkeypatch.setattr(claude_cli_tool, "WRAPPER_PATH", claude_cli_tool.Path("/tmp/hermes-call-claude")) + + def _fake_run(cmd, capture_output, text, timeout): + payload = { + "success": True, + "result": "OK", + "session_id": "abc123", + "command": cmd, + } + return subprocess.CompletedProcess(cmd, 0, stdout=json.dumps(payload), stderr="") + + monkeypatch.setattr(claude_cli_tool.subprocess, "run", _fake_run) + result = json.loads( + claude_cli_tool.claude_cli_run_tool( + prompt="Reply with exactly: OK", + workdir="/tmp", + allowed_tools=["Read", "Bash(git *)"], + timeout_seconds=30, + ) + ) + assert result["success"] is True + assert result["result"] == "OK" + assert "--allowed-tools" in result["command"] + + +def test_claude_cli_run_tool_returns_error_on_non_json(monkeypatch): + monkeypatch.setattr(claude_cli_tool, "BRIDGE_SCRIPT_PATH", claude_cli_tool.Path("/tmp/bridge.py")) + monkeypatch.setattr(claude_cli_tool, "_path_exists", lambda _path: True) + monkeypatch.setattr(claude_cli_tool.shutil, "which", lambda name: "/usr/bin/claude" if name == "claude" else None) + monkeypatch.setattr(claude_cli_tool, "WRAPPER_PATH", claude_cli_tool.Path("/tmp/hermes-call-claude")) + + def _fake_run(cmd, capture_output, text, timeout): + return subprocess.CompletedProcess(cmd, 1, stdout="not-json", stderr="boom") + + monkeypatch.setattr(claude_cli_tool.subprocess, "run", _fake_run) + result = json.loads(claude_cli_tool.claude_cli_run_tool(prompt="hi")) + assert "error" in result + assert "non-JSON" in result["error"] diff --git a/tools/claude_cli_tool.py b/tools/claude_cli_tool.py new file mode 100644 index 000000000..1cf542a5d --- /dev/null +++ b/tools/claude_cli_tool.py @@ -0,0 +1,169 @@ +#!/usr/bin/env python3 +"""Hermes tool for delegating one-shot work to the local Claude Code CLI.""" + +from __future__ import annotations + +import json +import os +import shutil +import subprocess +import sys +from pathlib import Path +from typing import Any + +from tools.registry import registry, tool_error + + +BRIDGE_SCRIPT_PATH = Path.home() / ".hermes" / "hermes-agent" / "scripts" / "run_claude_cli_json.py" +WRAPPER_PATH = Path.home() / ".local" / "bin" / "hermes-call-claude" +DEFAULT_MODEL = "opus" +DEFAULT_MAX_TURNS = 10 +DEFAULT_TIMEOUT_SECONDS = 300 + + +def _path_exists(path: Path) -> bool: + return path.exists() + + +def check_claude_cli_requirements() -> bool: + return shutil.which("claude") is not None and _path_exists(BRIDGE_SCRIPT_PATH) + + +def _coerce_allowed_tools(value: Any) -> str | None: + if value is None: + return None + if isinstance(value, list): + joined = ",".join(str(item).strip() for item in value if str(item).strip()) + return joined or None + text = str(value).strip() + return text or None + + +def claude_cli_run_tool( + *, + prompt: str, + workdir: str | None = None, + model: str = DEFAULT_MODEL, + max_turns: int = DEFAULT_MAX_TURNS, + allowed_tools: Any = None, + append_system_prompt: str | None = None, + timeout_seconds: int = DEFAULT_TIMEOUT_SECONDS, +) -> str: + if not prompt or not prompt.strip(): + return tool_error("prompt is required") + if not _path_exists(BRIDGE_SCRIPT_PATH): + return tool_error(f"Claude CLI bridge script not found: {BRIDGE_SCRIPT_PATH}") + claude_path = shutil.which("claude") + if not claude_path: + return tool_error("claude CLI not found on PATH") + + launcher = str(WRAPPER_PATH if _path_exists(WRAPPER_PATH) else Path(sys.executable)) + command = [launcher] + if launcher == sys.executable: + command.append(str(BRIDGE_SCRIPT_PATH)) + command.append(prompt) + if workdir: + command.extend(["--workdir", str(Path(workdir).expanduser())]) + if model: + command.extend(["--model", model]) + command.extend(["--max-turns", str(max_turns)]) + normalized_allowed = _coerce_allowed_tools(allowed_tools) + if normalized_allowed: + command.extend(["--allowed-tools", normalized_allowed]) + if append_system_prompt: + command.extend(["--append-system-prompt", append_system_prompt]) + command.extend(["--timeout", str(timeout_seconds)]) + + try: + completed = subprocess.run( + command, + capture_output=True, + text=True, + timeout=timeout_seconds + 5, + ) + except subprocess.TimeoutExpired: + return tool_error(f"Claude CLI bridge timed out after {timeout_seconds + 5}s") + + stdout = (completed.stdout or "").strip() + stderr = (completed.stderr or "").strip() + if not stdout: + return tool_error(stderr or "Claude CLI bridge produced no output") + try: + payload = json.loads(stdout) + except json.JSONDecodeError: + return tool_error(f"Claude CLI bridge returned non-JSON output: {stdout[:400]}") + + if stderr and not payload.get("stderr"): + payload["stderr"] = stderr + return json.dumps(payload, ensure_ascii=False) + + +CLAUDE_CLI_RUN_SCHEMA = { + "name": "claude_cli_run", + "description": ( + "Run the local Claude Code CLI in print mode and return structured JSON. " + "Use this when you specifically want the first-party Claude Code execution path " + "instead of Hermes's native Anthropic provider, for example when local Claude Code " + "auth works but native Anthropic provider billing/auth semantics do not." + ), + "parameters": { + "type": "object", + "properties": { + "prompt": { + "type": "string", + "description": "Prompt to send to Claude Code CLI.", + }, + "workdir": { + "type": "string", + "description": "Optional working directory for the claude process.", + }, + "model": { + "type": "string", + "description": "Claude model alias/name (default: opus).", + "default": DEFAULT_MODEL, + }, + "max_turns": { + "type": "integer", + "description": "Max turns for claude -p (default: 10).", + "default": DEFAULT_MAX_TURNS, + "minimum": 1, + }, + "allowed_tools": { + "oneOf": [ + {"type": "string"}, + {"type": "array", "items": {"type": "string"}}, + ], + "description": "Optional pass-through for Claude Code --allowedTools.", + }, + "append_system_prompt": { + "type": "string", + "description": "Optional additional system prompt passed to Claude Code.", + }, + "timeout_seconds": { + "type": "integer", + "description": "Subprocess timeout in seconds (default: 300).", + "default": DEFAULT_TIMEOUT_SECONDS, + "minimum": 1, + }, + }, + "required": ["prompt"], + }, +} + + +registry.register( + name="claude_cli_run", + toolset="terminal", + schema=CLAUDE_CLI_RUN_SCHEMA, + handler=lambda args, **kw: claude_cli_run_tool( + prompt=args.get("prompt", ""), + workdir=args.get("workdir"), + model=args.get("model", DEFAULT_MODEL), + max_turns=args.get("max_turns", DEFAULT_MAX_TURNS), + allowed_tools=args.get("allowed_tools"), + append_system_prompt=args.get("append_system_prompt"), + timeout_seconds=args.get("timeout_seconds", DEFAULT_TIMEOUT_SECONDS), + ), + check_fn=check_claude_cli_requirements, + emoji="🧠", +)