From 27d2e17d3aa1b66b682a1721704c1cd71fcfd140 Mon Sep 17 00:00:00 2001 From: Jer Miller Date: Sun, 8 Feb 2026 11:33:42 -0700 Subject: [PATCH] Replace Anthropic run_tools() with Claude CLI subprocess MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Swap the SDK-based MCP tool loop in think/providers/anthropic.py for a CLIRunner-based implementation that spawns `claude` as a subprocess. - Add _translate_claude() to map Claude CLI JSONL events (system, assistant, user, result) to our standard Event format - Replace run_tools() body: assemble_prompt → CLIRunner → finish event - Remove dead code: ToolExecutor, _get_mcp_tools, _emit_thinking_event, _MAX_TOOL_ITERATIONS - Clean imports: remove json, ToolParam, ToolUseBlock, create_mcp_client, ThinkingEvent, extract_tool_result; add cli.py imports - Update tests to mock CLIRunner instead of SDK; remove unused _setup_fastmcp_stub - Add _translate_claude unit tests and JSONL fixture for tool-use cycle run_generate and run_agenerate are unchanged. --- tests/fixtures/claude_cli_events.jsonl | 6 + tests/test_anthropic.py | 142 +++++--- tests/test_anthropic_cli.py | 329 ++++++++++++++++++ think/providers/anthropic.py | 452 +++++++++---------------- 4 files changed, 586 insertions(+), 343 deletions(-) create mode 100644 tests/fixtures/claude_cli_events.jsonl create mode 100644 tests/test_anthropic_cli.py diff --git a/tests/fixtures/claude_cli_events.jsonl b/tests/fixtures/claude_cli_events.jsonl new file mode 100644 index 000000000..f00a89e42 --- /dev/null +++ b/tests/fixtures/claude_cli_events.jsonl @@ -0,0 +1,6 @@ +{"type":"system","subtype":"init","session_id":"test-session-abc123","model":"claude-sonnet-4-20250514","tools":[{"name":"Read","type":"computer"}],"permissionMode":"plan"} +{"type":"assistant","message":{"id":"msg_01ABC","role":"assistant","content":[{"type":"text","text":"I'll read the file for you."}],"model":"claude-sonnet-4-20250514","stop_reason":null},"session_id":"test-session-abc123"} +{"type":"assistant","message":{"id":"msg_01ABC","role":"assistant","content":[{"type":"tool_use","id":"toolu_01XYZ","name":"Read","input":{"file_path":"/tmp/test.txt"}}],"model":"claude-sonnet-4-20250514","stop_reason":null},"session_id":"test-session-abc123"} +{"type":"user","message":{"role":"user","content":[{"tool_use_id":"toolu_01XYZ","type":"tool_result","content":"Hello, world!"}]}} +{"type":"assistant","message":{"id":"msg_02DEF","role":"assistant","content":[{"type":"text","text":"The file contains: Hello, world!"}],"model":"claude-sonnet-4-20250514","stop_reason":"end_turn"},"session_id":"test-session-abc123"} +{"type":"result","total_cost_usd":0.0042,"usage":{"input_tokens":150,"output_tokens":30},"session_id":"test-session-abc123","duration_ms":1234} diff --git a/tests/test_anthropic.py b/tests/test_anthropic.py index 4d76f7a87..20df6b224 100644 --- a/tests/test_anthropic.py +++ b/tests/test_anthropic.py @@ -117,50 +117,96 @@ def _setup_anthropic_stub( sys.modules["anthropic.types"] = anthropic_types_stub -def _setup_fastmcp_stub(monkeypatch): - """Mock fastmcp client.""" - fastmcp_stub = types.ModuleType("fastmcp") - fastmcp_fastmcp_stub = types.ModuleType("fastmcp.fastmcp") - - class DummyClient: - def __init__(self, *a, **k): - pass - - async def __aenter__(self): - return self - - async def __aexit__(self, *a): - pass - - async def list_tools(self): - return [] - - async def call_tool(self, name, arguments): - return SimpleNamespace( - content=[SimpleNamespace(type="text", text=f"Called {name}")] +def _setup_claude_cli_stub( + monkeypatch, + provider_mod, + *, + error=False, + with_thinking=False, + with_redacted_thinking=False, +): + monkeypatch.setattr( + provider_mod, "check_cli_binary", lambda _name: "/usr/bin/claude" + ) + monkeypatch.setattr(provider_mod, "lookup_cli_session_id", lambda _agent_id: None) + + class DummyCLIRunner: + def __init__( + self, + cmd, + prompt_text, + translate, + callback, + aggregator, + cwd=None, + env=None, + timeout=600, + ): + self.translate = translate + self.callback = callback + self.aggregator = aggregator + self.cli_session_id = None + + async def run(self): + if error: + raise RuntimeError("boo") + + raw_events = [ + { + "type": "system", + "subtype": "init", + "session_id": "test-session-abc123", + } + ] + if with_thinking: + raw_events.append( + { + "type": "assistant", + "message": { + "content": [ + { + "type": "thinking", + "thinking": "I'm thinking about this...", + } + ] + }, + } + ) + if with_redacted_thinking: + raw_events.append( + { + "type": "assistant", + "message": { + "content": [{"type": "thinking", "thinking": "[redacted]"}] + }, + } + ) + raw_events.append( + { + "type": "assistant", + "message": {"content": [{"type": "text", "text": "ok"}]}, + } ) - fastmcp_fastmcp_stub.FastMCP = DummyClient + for raw_event in raw_events: + session_id = self.translate(raw_event, self.aggregator, self.callback) + if session_id: + self.cli_session_id = session_id - if "fastmcp" in sys.modules: - sys.modules.pop("fastmcp") - if "fastmcp.fastmcp" in sys.modules: - sys.modules.pop("fastmcp.fastmcp") - sys.modules["fastmcp"] = fastmcp_stub - sys.modules["fastmcp.fastmcp"] = fastmcp_fastmcp_stub + result = self.aggregator.flush_as_result() + return result or "Done." - def mock_create_mcp_client(_url=None): - return DummyClient() - - monkeypatch.setattr("think.utils.create_mcp_client", mock_create_mcp_client) + monkeypatch.setattr(provider_mod, "CLIRunner", DummyCLIRunner) def test_claude_main(monkeypatch, tmp_path, capsys): _setup_anthropic_stub(monkeypatch) - _setup_fastmcp_stub(monkeypatch) install_agents_stub() sys.modules.pop("think.providers.anthropic", None) - importlib.reload(importlib.import_module("think.providers.anthropic")) + provider_mod = importlib.reload( + importlib.import_module("think.providers.anthropic") + ) + _setup_claude_cli_stub(monkeypatch, provider_mod) mod = importlib.reload(importlib.import_module("think.agents")) journal = tmp_path / "journal" @@ -200,10 +246,12 @@ def test_claude_main(monkeypatch, tmp_path, capsys): def test_claude_outfile(monkeypatch, tmp_path, capsys): _setup_anthropic_stub(monkeypatch) - _setup_fastmcp_stub(monkeypatch) install_agents_stub() sys.modules.pop("think.providers.anthropic", None) - importlib.reload(importlib.import_module("think.providers.anthropic")) + provider_mod = importlib.reload( + importlib.import_module("think.providers.anthropic") + ) + _setup_claude_cli_stub(monkeypatch, provider_mod) mod = importlib.reload(importlib.import_module("think.agents")) journal = tmp_path / "journal" @@ -247,10 +295,12 @@ def test_claude_thinking_events(monkeypatch, tmp_path, capsys): """Test that thinking events are properly emitted for Claude models.""" # Setup anthropic stub with thinking _setup_anthropic_stub(monkeypatch, with_thinking=True) - _setup_fastmcp_stub(monkeypatch) install_agents_stub() sys.modules.pop("think.providers.anthropic", None) - importlib.reload(importlib.import_module("think.providers.anthropic")) + provider_mod = importlib.reload( + importlib.import_module("think.providers.anthropic") + ) + _setup_claude_cli_stub(monkeypatch, provider_mod, with_thinking=True) mod = importlib.reload(importlib.import_module("think.agents")) journal = tmp_path / "journal" @@ -279,8 +329,6 @@ def test_claude_thinking_events(monkeypatch, tmp_path, capsys): thinking_events = [e for e in events if e.get("event") == "thinking"] assert len(thinking_events) == 1 assert "I'm thinking about this..." in thinking_events[0]["summary"] - # Verify signature is captured - assert thinking_events[0].get("signature") == "test-signature-123" # Check that regular events are still present assert events[0]["event"] == "start" @@ -291,10 +339,12 @@ def test_claude_thinking_events(monkeypatch, tmp_path, capsys): def test_claude_redacted_thinking_events(monkeypatch, tmp_path, capsys): """Test that redacted thinking events are properly handled.""" _setup_anthropic_stub(monkeypatch, with_redacted_thinking=True) - _setup_fastmcp_stub(monkeypatch) install_agents_stub() sys.modules.pop("think.providers.anthropic", None) - importlib.reload(importlib.import_module("think.providers.anthropic")) + provider_mod = importlib.reload( + importlib.import_module("think.providers.anthropic") + ) + _setup_claude_cli_stub(monkeypatch, provider_mod, with_redacted_thinking=True) mod = importlib.reload(importlib.import_module("think.agents")) journal = tmp_path / "journal" @@ -323,8 +373,6 @@ def test_claude_redacted_thinking_events(monkeypatch, tmp_path, capsys): thinking_events = [e for e in events if e.get("event") == "thinking"] assert len(thinking_events) == 1 assert thinking_events[0]["summary"] == "[redacted]" - # Verify redacted_data is captured - assert thinking_events[0].get("redacted_data") == "encrypted-data-xyz" # Check that regular events are still present assert events[0]["event"] == "start" @@ -333,10 +381,12 @@ def test_claude_redacted_thinking_events(monkeypatch, tmp_path, capsys): def test_claude_outfile_error(monkeypatch, tmp_path, capsys): _setup_anthropic_stub(monkeypatch, error=True) - _setup_fastmcp_stub(monkeypatch) install_agents_stub() sys.modules.pop("think.providers.anthropic", None) - importlib.reload(importlib.import_module("think.providers.anthropic")) + provider_mod = importlib.reload( + importlib.import_module("think.providers.anthropic") + ) + _setup_claude_cli_stub(monkeypatch, provider_mod, error=True) mod = importlib.reload(importlib.import_module("think.agents")) journal = tmp_path / "journal" diff --git a/tests/test_anthropic_cli.py b/tests/test_anthropic_cli.py new file mode 100644 index 000000000..baee13aa2 --- /dev/null +++ b/tests/test_anthropic_cli.py @@ -0,0 +1,329 @@ +# SPDX-License-Identifier: AGPL-3.0-only +# Copyright (c) 2026 sol pbc + +"""Tests for Claude CLI translator in think.providers.anthropic.""" + +import json +from pathlib import Path + +import pytest + +from think.providers.anthropic import _translate_claude +from think.providers.cli import ThinkingAggregator +from think.providers.shared import JSONEventCallback + + +@pytest.fixture +def callback_events(): + """Collect emitted events.""" + events = [] + return events, JSONEventCallback(lambda e: events.append(e)) + + +@pytest.fixture +def aggregator(callback_events): + """Create a ThinkingAggregator.""" + _, cb = callback_events + return ThinkingAggregator(cb, model="claude-sonnet-4-20250514") + + +@pytest.fixture +def state(): + """Create mutable state dicts for translator.""" + return {"pending_tools": {}, "result_meta": {}} + + +class TestTranslateClaudeSystemEvent: + def test_init_returns_session_id(self, aggregator, callback_events, state): + events, cb = callback_events + event = { + "type": "system", + "subtype": "init", + "session_id": "sess-123", + "model": "claude-sonnet-4-20250514", + } + result = _translate_claude( + event, aggregator, cb, state["pending_tools"], state["result_meta"] + ) + assert result == "sess-123" + assert len(events) == 0 # No events emitted for system init + + def test_non_init_system_event(self, aggregator, callback_events, state): + events, cb = callback_events + event = {"type": "system", "subtype": "other"} + result = _translate_claude( + event, aggregator, cb, state["pending_tools"], state["result_meta"] + ) + assert result is None + + +class TestTranslateClaudeAssistantEvent: + def test_text_accumulates(self, aggregator, callback_events, state): + events, cb = callback_events + event = { + "type": "assistant", + "message": { + "id": "msg_01", + "content": [{"type": "text", "text": "Hello there"}], + }, + } + _translate_claude( + event, aggregator, cb, state["pending_tools"], state["result_meta"] + ) + assert aggregator.flush_as_result() == "Hello there" + assert len(events) == 0 # No events emitted for plain text + + def test_thinking_emits_event(self, aggregator, callback_events, state): + events, cb = callback_events + event = { + "type": "assistant", + "message": { + "id": "msg_01", + "content": [{"type": "thinking", "thinking": "Let me consider..."}], + }, + } + _translate_claude( + event, aggregator, cb, state["pending_tools"], state["result_meta"] + ) + assert len(events) == 1 + assert events[0]["event"] == "thinking" + assert events[0]["summary"] == "Let me consider..." + assert events[0]["model"] == "claude-sonnet-4-20250514" + assert events[0]["raw"] == [event] + + def test_tool_use_emits_tool_start(self, aggregator, callback_events, state): + events, cb = callback_events + event = { + "type": "assistant", + "message": { + "id": "msg_01", + "content": [ + { + "type": "tool_use", + "id": "toolu_01", + "name": "Read", + "input": {"file_path": "/tmp/test.txt"}, + } + ], + }, + } + _translate_claude( + event, aggregator, cb, state["pending_tools"], state["result_meta"] + ) + assert len(events) == 1 + assert events[0]["event"] == "tool_start" + assert events[0]["tool"] == "Read" + assert events[0]["args"] == {"file_path": "/tmp/test.txt"} + assert events[0]["call_id"] == "toolu_01" + assert events[0]["raw"] == [event] + # Verify pending_tools tracking + assert "toolu_01" in state["pending_tools"] + + def test_text_before_tool_use_flushed_as_thinking( + self, aggregator, callback_events, state + ): + """Text followed by tool_use should flush text as thinking.""" + events, cb = callback_events + event = { + "type": "assistant", + "message": { + "id": "msg_01", + "content": [ + {"type": "text", "text": "I'll read that file."}, + { + "type": "tool_use", + "id": "toolu_01", + "name": "Read", + "input": {"file_path": "/tmp/test.txt"}, + }, + ], + }, + } + _translate_claude( + event, aggregator, cb, state["pending_tools"], state["result_meta"] + ) + # Should emit: thinking (flushed text), then tool_start + assert len(events) == 2 + assert events[0]["event"] == "thinking" + assert events[0]["summary"] == "I'll read that file." + assert events[1]["event"] == "tool_start" + assert events[1]["tool"] == "Read" + + def test_multiple_tool_uses(self, aggregator, callback_events, state): + """Multiple tool_use blocks should each emit tool_start.""" + events, cb = callback_events + event = { + "type": "assistant", + "message": { + "id": "msg_01", + "content": [ + { + "type": "tool_use", + "id": "toolu_01", + "name": "Read", + "input": {"file_path": "/a.txt"}, + }, + { + "type": "tool_use", + "id": "toolu_02", + "name": "Read", + "input": {"file_path": "/b.txt"}, + }, + ], + }, + } + _translate_claude( + event, aggregator, cb, state["pending_tools"], state["result_meta"] + ) + assert len(events) == 2 + assert events[0]["event"] == "tool_start" + assert events[0]["call_id"] == "toolu_01" + assert events[1]["event"] == "tool_start" + assert events[1]["call_id"] == "toolu_02" + + +class TestTranslateClaudeUserEvent: + def test_tool_result_emits_tool_end(self, aggregator, callback_events, state): + events, cb = callback_events + # First, register a pending tool + state["pending_tools"]["toolu_01"] = { + "tool": "Read", + "args": {"file_path": "/tmp/test.txt"}, + } + event = { + "type": "user", + "message": { + "content": [ + { + "type": "tool_result", + "tool_use_id": "toolu_01", + "content": "file contents here", + } + ] + }, + } + _translate_claude( + event, aggregator, cb, state["pending_tools"], state["result_meta"] + ) + assert len(events) == 1 + assert events[0]["event"] == "tool_end" + assert events[0]["tool"] == "Read" + assert events[0]["args"] == {"file_path": "/tmp/test.txt"} + assert events[0]["result"] == "file contents here" + assert events[0]["call_id"] == "toolu_01" + assert events[0]["raw"] == [event] + # Verify pending tool was consumed + assert "toolu_01" not in state["pending_tools"] + + def test_tool_result_unknown_id(self, aggregator, callback_events, state): + """tool_result with unknown ID should still emit tool_end.""" + events, cb = callback_events + event = { + "type": "user", + "message": { + "content": [ + { + "type": "tool_result", + "tool_use_id": "toolu_unknown", + "content": "result", + } + ] + }, + } + _translate_claude( + event, aggregator, cb, state["pending_tools"], state["result_meta"] + ) + assert len(events) == 1 + assert events[0]["event"] == "tool_end" + assert events[0]["tool"] == "" + assert events[0]["call_id"] == "toolu_unknown" + + +class TestTranslateClaudeResultEvent: + def test_stores_usage_and_cost(self, aggregator, callback_events, state): + events, cb = callback_events + event = { + "type": "result", + "total_cost_usd": 0.0042, + "usage": {"input_tokens": 150, "output_tokens": 30}, + "session_id": "sess-123", + } + _translate_claude( + event, aggregator, cb, state["pending_tools"], state["result_meta"] + ) + assert state["result_meta"]["cost_usd"] == 0.0042 + assert state["result_meta"]["usage"]["input_tokens"] == 150 + assert state["result_meta"]["usage"]["output_tokens"] == 30 + assert state["result_meta"]["usage"]["total_tokens"] == 180 + assert len(events) == 0 # No events emitted for result + + def test_result_without_usage(self, aggregator, callback_events, state): + events, cb = callback_events + event = {"type": "result", "total_cost_usd": 0.001} + _translate_claude( + event, aggregator, cb, state["pending_tools"], state["result_meta"] + ) + assert state["result_meta"]["cost_usd"] == 0.001 + assert "usage" not in state["result_meta"] + + +class TestTranslateClaudeUnknownEvent: + def test_unknown_type_ignored(self, aggregator, callback_events, state): + events, cb = callback_events + event = {"type": "unknown_event", "data": "whatever"} + result = _translate_claude( + event, aggregator, cb, state["pending_tools"], state["result_meta"] + ) + assert result is None + assert len(events) == 0 + + +class TestTranslateClaudeFixtureSequence: + """Test processing a full sequence of events from fixture file.""" + + def test_full_tool_cycle(self, callback_events): + """Process a complete tool-use cycle and verify all events.""" + events, cb = callback_events + aggregator = ThinkingAggregator(cb, model="claude-sonnet-4-20250514") + pending_tools: dict = {} + result_meta: dict = {} + + fixture_path = Path(__file__).parent / "fixtures" / "claude_cli_events.jsonl" + with open(fixture_path) as f: + raw_events = [json.loads(line) for line in f if line.strip()] + + session_id = None + for raw_event in raw_events: + sid = _translate_claude( + raw_event, aggregator, cb, pending_tools, result_meta + ) + if sid: + session_id = sid + + # Verify session ID captured from init event + assert session_id == "test-session-abc123" + + # Verify events emitted + event_types = [e["event"] for e in events] + assert "tool_start" in event_types + assert "tool_end" in event_types + + # Verify tool pairing + tool_starts = [e for e in events if e["event"] == "tool_start"] + tool_ends = [e for e in events if e["event"] == "tool_end"] + assert len(tool_starts) == 1 + assert len(tool_ends) == 1 + assert tool_starts[0]["call_id"] == tool_ends[0]["call_id"] + assert tool_ends[0]["result"] == "Hello, world!" + + # Verify usage captured + assert result_meta["cost_usd"] == 0.0042 + assert result_meta["usage"]["input_tokens"] == 150 + assert result_meta["usage"]["output_tokens"] == 30 + + # Verify final text in aggregator (last assistant text) + result = aggregator.flush_as_result() + assert "Hello, world!" in result + + # Verify all pending tools consumed + assert len(pending_tools) == 0 diff --git a/think/providers/anthropic.py b/think/providers/anthropic.py index 3ef17f7fd..e991836f5 100644 --- a/think/providers/anthropic.py +++ b/think/providers/anthropic.py @@ -31,7 +31,6 @@ timeout_s : float, optional from __future__ import annotations -import json import logging import os import traceback @@ -42,18 +41,21 @@ from anthropic.types import ( MessageParam, RedactedThinkingBlock, ThinkingBlock, - ToolParam, - ToolUseBlock, ) from think.models import CLAUDE_SONNET_4 -from think.utils import create_mcp_client, now_ms - +from think.utils import now_ms + +from .cli import ( + CLIRunner, + ThinkingAggregator, + assemble_prompt, + check_cli_binary, + lookup_cli_session_id, +) from .shared import ( GenerateResult, JSONEventCallback, - ThinkingEvent, - extract_tool_result, ) # Default values are now handled internally @@ -104,338 +106,194 @@ def _resolve_agent_thinking_params( return _compute_thinking_params(max_output_tokens) -def _emit_thinking_event( - block: ThinkingBlock | RedactedThinkingBlock, - model: str, +def _translate_claude( + event: dict[str, Any], + aggregator: ThinkingAggregator, callback: JSONEventCallback, -) -> None: - """Emit a thinking event for a ThinkingBlock or RedactedThinkingBlock.""" - if isinstance(block, ThinkingBlock): - thinking_event: ThinkingEvent = { - "event": "thinking", - "ts": now_ms(), - "summary": block.thinking, - "model": model, - "signature": block.signature, - } - callback.emit(thinking_event) - elif isinstance(block, RedactedThinkingBlock): - redacted_event: ThinkingEvent = { - "event": "thinking", - "ts": now_ms(), - "summary": "[redacted]", - "model": model, - "redacted_data": block.data, - } - callback.emit(redacted_event) + pending_tools: dict[str, dict[str, Any]], + result_meta: dict[str, Any], +) -> str | None: + """Translate a Claude CLI JSONL event into our Event format. + Args: + event: Raw parsed JSON event from Claude CLI stdout. + aggregator: ThinkingAggregator for text buffering. + callback: JSONEventCallback for emitting events. + pending_tools: Mutable dict tracking active tool calls (tool_use_id -> {tool, args}). + result_meta: Mutable dict for storing cost/usage from result event. -_MAX_TOOL_ITERATIONS = 25 # Safety limit for agentic loop iterations - + Returns: + Session ID string from init events, None otherwise. + """ + event_type = event.get("type") + + if event_type == "system": + if event.get("subtype") == "init": + return event.get("session_id") + + elif event_type == "assistant": + message = event.get("message", {}) + content_blocks = message.get("content", []) + + # Two-pass: text/thinking first, then tool_use + tool_use_blocks = [] + for block in content_blocks: + block_type = block.get("type") + if block_type == "text": + aggregator.accumulate(block.get("text", "")) + elif block_type == "thinking": + thinking_event: dict[str, Any] = { + "event": "thinking", + "summary": block.get("thinking", ""), + "raw": [event], + } + if aggregator._model: + thinking_event["model"] = aggregator._model + callback.emit(thinking_event) + elif block_type == "tool_use": + tool_use_blocks.append(block) -class ToolExecutor: - """Handle MCP tool execution and result formatting for Anthropic.""" + for block in tool_use_blocks: + aggregator.flush_as_thinking(raw_events=[event]) - def __init__( - self, - mcp_client: Any, - callback: JSONEventCallback, - agent_id: str | None = None, - name: str | None = None, - day: str | None = None, - ) -> None: - self.mcp = mcp_client - self.callback = callback - self.agent_id = agent_id - self.name = name - self.day = day + tool_id = block.get("id", "") + tool_name = block.get("name", "") + tool_args = block.get("input", {}) - async def execute_tool(self, tool_use: ToolUseBlock) -> dict: - """Execute ``tool_use`` and return a Claude ``tool_result`` block.""" - call_id = tool_use.id # Use Claude's tool_use_id as call_id - self.callback.emit( - { - "event": "tool_start", - "tool": tool_use.name, - "args": tool_use.input, - "call_id": call_id, - } - ) + pending_tools[tool_id] = {"tool": tool_name, "args": tool_args} - # Build _meta dict for passing agent identity and context - meta = {} - if self.agent_id: - meta["agent_id"] = self.agent_id - if self.name: - meta["name"] = self.name - if self.day: - meta["day"] = self.day - - try: - try: - result = await self.mcp.session.call_tool( - name=tool_use.name, - arguments=tool_use.input, - meta=meta, - ) - except RuntimeError: - await self.mcp.__aenter__() - result = await self.mcp.session.call_tool( - name=tool_use.name, - arguments=tool_use.input, - meta=meta, - ) - result_data = extract_tool_result(result) - self.callback.emit( - { - "event": "tool_end", - "tool": tool_use.name, - "args": tool_use.input, - "result": result_data, - "call_id": call_id, - } - ) - content = ( - result_data if isinstance(result_data, str) else json.dumps(result_data) - ) - except Exception as exc: # pragma: no cover - unexpected - self.callback.emit( + callback.emit( { - "event": "tool_end", - "tool": tool_use.name, - "args": tool_use.input, - "result": {"error": str(exc)}, - "call_id": call_id, + "event": "tool_start", + "tool": tool_name, + "args": tool_args, + "call_id": tool_id, + "raw": [event], } ) - content = f"Error: {exc}" - - return { - "type": "tool_result", - "tool_use_id": tool_use.id, - "content": content, - } - - -async def _get_mcp_tools( - mcp: Any, allowed_tools: list[str] | None = None -) -> list[ToolParam]: - """Return a list of MCP tools formatted for Claude using ``mcp``. - - Args: - mcp: MCP client instance - allowed_tools: Optional list of allowed tool names to filter - """ - if not hasattr(mcp, "list_tools"): - return [] - - tools = [] - tool_list = await mcp.list_tools() - - for tool in tool_list: - # Filter by allowed tools if specified - if allowed_tools and tool.name not in allowed_tools: - continue + elif event_type == "user": + message = event.get("message", {}) + content_blocks = message.get("content", []) + + for block in content_blocks: + if block.get("type") == "tool_result": + tool_use_id = block.get("tool_use_id", "") + tool_info = pending_tools.pop(tool_use_id, {}) + + callback.emit( + { + "event": "tool_end", + "tool": tool_info.get("tool", ""), + "args": tool_info.get("args"), + "result": block.get("content", ""), + "call_id": tool_use_id, + "raw": [event], + } + ) - tools.append( - { - "name": tool.name, - "description": tool.description or "", - "input_schema": tool.inputSchema - or {"type": "object", "properties": {}, "required": []}, + elif event_type == "result": + result_meta["cost_usd"] = event.get("total_cost_usd") + usage = event.get("usage") + if usage: + result_meta["usage"] = { + "input_tokens": usage.get("input_tokens"), + "output_tokens": usage.get("output_tokens"), + "total_tokens": ( + (usage.get("input_tokens") or 0) + (usage.get("output_tokens") or 0) + ), } - ) - return tools + return None async def run_tools( config: dict[str, Any], on_event: Callable[[dict], None] | None = None, ) -> str: - """Run a prompt with MCP tool-calling support via Anthropic Claude. + """Run a prompt with tool-calling support via Claude CLI subprocess. + + Spawns the Claude CLI in streaming JSON mode and translates its + JSONL output into our standard Event format. Args: config: Complete configuration dictionary including prompt, system_instruction, user_instruction, extra_context, model, etc. on_event: Optional event callback """ - # Extract config values directly - prompt = config.get("prompt", "") model = config.get("model", _DEFAULT_MODEL) - system_instruction = config.get("system_instruction") - user_instruction = config.get("user_instruction") - extra_context = config.get("extra_context") - transcript = config.get("transcript") - mcp_server_url = config.get("mcp_server_url") - tools_filter = config.get("tools") - max_output_tokens = config.get("max_output_tokens", _DEFAULT_MAX_TOKENS) - thinking_budget_config = config.get("thinking_budget") continue_from = config.get("continue_from") - agent_id = config.get("agent_id") - name = config.get("name") - day = config.get("day") callback = JSONEventCallback(on_event) try: - api_key = os.getenv("ANTHROPIC_API_KEY") - if not api_key: - raise RuntimeError("ANTHROPIC_API_KEY not set") - - client = AsyncAnthropic(api_key=api_key) + check_cli_binary("claude") + + prompt_body, system_instruction = assemble_prompt(config) + + cmd = [ + "claude", + "-p", + "-", + "--output-format", + "stream-json", + "--verbose", + "--permission-mode", + "plan", + "--model", + model, + ] + + if system_instruction: + cmd.extend(["--system-prompt", system_instruction]) - # Note: Start event is emitted by agents.py (unified event ownership) - - # Build initial messages - check for continuation first if continue_from: - # Load previous conversation history using shared function - from ..agents import parse_agent_events_to_turns - - messages = parse_agent_events_to_turns(continue_from) - # Add new prompt as continuation - messages.append({"role": "user", "content": prompt}) - else: - # Fresh conversation - messages: list[MessageParam] = [] - # Prepend transcript if provided (from day/segment input assembly) - if transcript: - messages.append({"role": "user", "content": transcript}) - if extra_context: - messages.append({"role": "user", "content": extra_context}) - if user_instruction: - messages.append({"role": "user", "content": user_instruction}) - messages.append({"role": "user", "content": prompt}) - - # Initialize tools and executor if MCP server URL provided - if mcp_server_url: - async with create_mcp_client(str(mcp_server_url)) as mcp: - if tools_filter and isinstance(tools_filter, list): - logger.info(f"Using tool filter with allowed tools: {tools_filter}") - - tools = await _get_mcp_tools(mcp, tools_filter) - tool_executor = ToolExecutor( - mcp, callback, agent_id=agent_id, name=name, day=day + session_id = lookup_cli_session_id(continue_from) + if session_id: + cmd.extend(["--resume", session_id]) + else: + logger.warning( + "No CLI session ID found for continue_from=%s", continue_from ) - thinking_budget, effective_max_tokens = _resolve_agent_thinking_params( - max_output_tokens, thinking_budget_config - ) - - for _ in range(_MAX_TOOL_ITERATIONS): - # Build request params - thinking always enabled - create_params = { - "model": model, - "max_tokens": effective_max_tokens, - "system": system_instruction, - "messages": messages, - "thinking": { - "type": "enabled", - "budget_tokens": thinking_budget, - }, - } - if tools: - create_params["tools"] = tools - - response = await client.messages.create(**create_params) - - tool_uses = [] - final_text = "" - for block in response.content: - if getattr(block, "type", None) == "text": - final_text += block.text - elif getattr(block, "type", None) == "tool_use": - tool_uses.append(block) - elif isinstance(block, (ThinkingBlock, RedactedThinkingBlock)): - _emit_thinking_event(block, model, callback) - - messages.append({"role": "assistant", "content": response.content}) - - if not tool_uses: - # Model is done - check for tool-only completion - tool_only = False - if not final_text: - final_text = "Done." - tool_only = True - logger.info( - "Tool-only completion, using synthetic response" - ) - finish_event = { - "event": "finish", - "result": final_text, - "usage": _extract_usage_dict(response), - "ts": now_ms(), - } - if tool_only: - finish_event["tool_only"] = True - finish_reason = _normalize_finish_reason( - getattr(response, "stop_reason", None) - ) - if finish_reason: - finish_event["reason"] = finish_reason - callback.emit(finish_event) - return final_text - - results = [] - for tool_use in tool_uses: - result = await tool_executor.execute_tool(tool_use) - results.append(result) - - messages.append({"role": "user", "content": results}) - else: - # Hit iteration limit - treat as tool-only completion - logger.warning( - f"Hit max iterations ({_MAX_TOOL_ITERATIONS}), completing" - ) - callback.emit( - { - "event": "finish", - "result": "Done.", - "tool_only": True, - "reason": "max_iterations", - "ts": now_ms(), - } - ) - return "Done." - else: - # No MCP tools - single response only - thinking_budget, effective_max_tokens = _resolve_agent_thinking_params( - max_output_tokens, thinking_budget_config - ) - create_params = { - "model": model, - "max_tokens": effective_max_tokens, - "system": system_instruction, - "messages": messages, - "thinking": { - "type": "enabled", - "budget_tokens": thinking_budget, - }, - } + aggregator = ThinkingAggregator(callback, model=model) + pending_tools: dict[str, dict[str, Any]] = {} + result_meta: dict[str, Any] = {} + + def translate( + event: dict[str, Any], + agg: ThinkingAggregator, + cb: JSONEventCallback, + ) -> str | None: + return _translate_claude(event, agg, cb, pending_tools, result_meta) + + runner = CLIRunner( + cmd=cmd, + prompt_text=prompt_body, + translate=translate, + callback=callback, + aggregator=aggregator, + ) - response = await client.messages.create(**create_params) + result = await runner.run() - final_text = "" - for block in response.content: - if getattr(block, "type", None) == "text": - final_text += block.text - elif isinstance(block, (ThinkingBlock, RedactedThinkingBlock)): - _emit_thinking_event(block, model, callback) + # Build finish event with usage from result meta + usage_dict = result_meta.get("usage") + cost_usd = result_meta.get("cost_usd") + if usage_dict and cost_usd is not None: + usage_dict["cost_usd"] = cost_usd - finish_event = { + callback.emit( + { "event": "finish", - "result": final_text, - "usage": _extract_usage_dict(response), + "result": result, + "cli_session_id": runner.cli_session_id, + "usage": usage_dict, "ts": now_ms(), } - finish_reason = _normalize_finish_reason( - getattr(response, "stop_reason", None) - ) - if finish_reason: - finish_event["reason"] = finish_reason - callback.emit(finish_event) - return final_text + ) + + return result except Exception as exc: callback.emit( { -- 2.51.2