claude_code

claude_code

ClaudeCodeAgent -- wraps the Claude Agent SDK via Node.js subprocess bridge.

Spawns a Node.js runner process that calls the @anthropic-ai/claude-code SDK, communicating via JSON over stdin/stdout with sentinel-delimited output.

The engine parameter is accepted for interface conformance with BaseAgent but is not used -- inference is handled entirely by the Claude Agent SDK.

Classes

ClaudeCodeAgent

ClaudeCodeAgent(
    engine: InferenceEngine,
    model: str,
    *,
    bus: Optional[EventBus] = None,
    temperature: Optional[float] = None,
    max_tokens: Optional[int] = None,
    api_key: str = "",
    workspace: str = "",
    session_id: str = "",
    allowed_tools: Optional[List[str]] = None,
    system_prompt: str = "",
    timeout: int = 300,
)

Bases: BaseAgent

Agent that wraps the Claude Agent SDK via a Node.js subprocess.

Spawns a Node.js process running dist/index.js which imports @anthropic-ai/claude-code and streams agentic responses. Results are communicated back via sentinel-delimited JSON on stdout.

The engine parameter is accepted for BaseAgent interface conformance but is not used -- all inference is handled by the Claude Agent SDK.

Source code in src/diapason/agents/claude_code.py

def __init__(
    self,
    engine: InferenceEngine,
    model: str,
    *,
    bus: Optional[EventBus] = None,
    temperature: Optional[float] = None,
    max_tokens: Optional[int] = None,
    api_key: str = "",
    workspace: str = "",
    session_id: str = "",
    allowed_tools: Optional[List[str]] = None,
    system_prompt: str = "",
    timeout: int = 300,
) -> None:
    super().__init__(
        engine,
        model,
        bus=bus,
        temperature=temperature,
        max_tokens=max_tokens,
    )
    self._api_key = api_key or os.environ.get("ANTHROPIC_API_KEY", "")
    self._workspace = workspace or os.getcwd()
    self._session_id = session_id
    self._allowed_tools = allowed_tools
    self._system_prompt = system_prompt
    self._timeout = timeout

Methods:

run

run(
    input: str,
    context: Optional[AgentContext] = None,
    **kwargs: Any,
) -> AgentResult

Execute a query via the Claude Agent SDK subprocess.

Spawns node dist/index.js, writes a JSON request to stdin, and reads sentinel-delimited JSON output from stdout.

Source code in src/diapason/agents/claude_code.py

def run(
    self,
    input: str,
    context: Optional[AgentContext] = None,
    **kwargs: Any,
) -> AgentResult:
    """Execute a query via the Claude Agent SDK subprocess.

    Spawns ``node dist/index.js``, writes a JSON request to stdin, and
    reads sentinel-delimited JSON output from stdout.
    """
    self._emit_turn_start(input)

    runner_dir = self._ensure_runner()

    # Build the request payload
    request = {
        "prompt": input,
        "api_key": self._api_key,
        "workspace": self._workspace,
        "allowed_tools": self._allowed_tools or [],
        "system_prompt": self._system_prompt,
        "session_id": self._session_id,
    }

    try:
        proc = subprocess.run(
            ["node", "dist/index.js"],
            cwd=str(runner_dir),
            input=json.dumps(request),
            capture_output=True,
            text=True,
            timeout=self._timeout,
        )
    except subprocess.TimeoutExpired:
        self._emit_turn_end(turns=1, error=True)
        return AgentResult(
            content=f"Claude Code agent timed out after {self._timeout}s.",
            turns=1,
            metadata={"error": True, "error_type": "timeout"},
        )

    if proc.returncode != 0:
        stderr = proc.stderr.strip() if proc.stderr else "Unknown error"
        logger.error(
            "claude_code_runner exited with code %d: %s",
            proc.returncode,
            stderr,
        )
        self._emit_turn_end(turns=1, error=True)
        return AgentResult(
            content=f"Claude Code agent failed: {stderr}",
            turns=1,
            metadata={"error": True, "returncode": proc.returncode},
        )

    # Parse sentinel-delimited output
    content, tool_results, metadata = self._parse_output(proc.stdout)

    self._emit_turn_end(turns=1)
    return AgentResult(
        content=content,
        tool_results=tool_results,
        turns=1,
        metadata=metadata,
    )

Functions: