diff options
Diffstat (limited to 'agent/orchestrator/models/claude_code.py')
| -rw-r--r-- | agent/orchestrator/models/claude_code.py | 68 |
1 files changed, 68 insertions, 0 deletions
diff --git a/agent/orchestrator/models/claude_code.py b/agent/orchestrator/models/claude_code.py new file mode 100644 index 0000000..c73df98 --- /dev/null +++ b/agent/orchestrator/models/claude_code.py @@ -0,0 +1,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) |
