aboutsummaryrefslogtreecommitdiff
path: root/agent/orchestrator/models/claude_code.py
blob: c73df98543ab4f525b811485aefbb174453b27cb (plain)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
"""Claude Code backend.

Shells out to `claude -p` for non-interactive turns. Claude Code handles its
own persona discovery (CLAUDE.md / SOUL.md from the workspace) and its own MCP
tool dispatch internally; this backend just formats the prompt and parses
the response.
"""

import os
import subprocess
from pathlib import Path

from .base import ModelBackend, ModelRequest, ModelResponse


_ROLE_LABELS = {
    "user": "User",
    "assistant": "Assistant",
    "tool": "Tool",
    "system": "System",
}


class ClaudeCodeBackend(ModelBackend):
    def __init__(
        self,
        workspace: Path = Path(os.environ.get("SENIA_WORKSPACE", "/workspace")),
        timeout: int = 300,
        allowed_tools: str = "mcp__senia__*",
    ):
        self.workspace = workspace
        self.timeout = timeout
        self.allowed_tools = allowed_tools

    def call(self, request: ModelRequest) -> ModelResponse:
        prompt = self._format_prompt(request)
        result = subprocess.run(
            [
                "claude", "-p", prompt,
                "--output-format", "text",
                "--allowedTools", self.allowed_tools,
            ],
            cwd=self.workspace,
            capture_output=True,
            text=True,
            timeout=self.timeout,
        )
        if result.returncode != 0:
            raise RuntimeError(
                f"claude exited {result.returncode}: {result.stderr.strip()}"
            )
        return ModelResponse(content=result.stdout.strip())

    def _format_prompt(self, request: ModelRequest) -> str:
        """Flatten messages + dynamic context into a single prompt string.

        Claude Code's -p mode takes one big prompt, not a structured
        conversation. We render the conversation as a readable transcript with
        the per-turn context prepended. Persona is NOT here; Claude Code
        finds it in the workspace itself.
        """
        parts = []
        if request.turn_context:
            parts.append(f"## Context for this turn\n\n{request.turn_context}\n")
        for msg in request.messages:
            parts.append(f"## {_ROLE_LABELS[msg.role]}\n\n{msg.content}\n")
        parts.append("## Assistant\n\n")
        return "\n".join(parts)