144 lines
5.3 KiB
Python
144 lines
5.3 KiB
Python
"""Claude agent definition, backed by the Claude Agent SDK.
|
|
|
|
``system_prompt`` is used verbatim, or per-kind granules from ``prompts``
|
|
are assembled via ``agents/prompts.py``. ``skill_sets`` are
|
|
``<skill>/SKILL.md`` directories, each becoming a local SDK plugin.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from collections.abc import Mapping # noqa: TC003
|
|
from pathlib import Path # noqa: TC003
|
|
|
|
from pydantic import BaseModel, ConfigDict, Field, model_validator
|
|
|
|
from beaver_gateway.agents.base import BaseAgent
|
|
from beaver_gateway.agents.policy import PolicyRule # noqa: TC001
|
|
from beaver_gateway.agents.prompts import PromptSource # noqa: TC001
|
|
from beaver_gateway.conversations.kinds import KINDS, Kind
|
|
|
|
__all__ = ["PROGRESS_MODELS", "ClaudeAgent", "ClaudeOptions", "Prompts", "SkillSets"]
|
|
|
|
PROGRESS_MODELS = frozenset({"claude-opus-5-5", "claude-fable-5-1"})
|
|
"""Models whose non-empty thinking is the text written between tool calls."""
|
|
|
|
|
|
class ClaudeOptions(BaseModel):
|
|
model_config = ConfigDict(frozen=True, arbitrary_types_allowed=True)
|
|
|
|
effort: str | None = None
|
|
thinking: Mapping[str, str] | None = None
|
|
"""Thinking config for the SDK; ``None`` keeps the CLI default."""
|
|
|
|
include_partial_messages: bool = True
|
|
"""Token-level deltas on the wire; off means one delta per finished block."""
|
|
|
|
permission_mode: str = "bypassPermissions"
|
|
tools: tuple[str, ...] | None = None
|
|
"""Base set of built-in tools; ``None`` keeps the CLI default set."""
|
|
|
|
disallowed_tools: tuple[str, ...] = ()
|
|
add_dirs: tuple[str, ...] = ()
|
|
env: Mapping[str, str] = Field(default_factory=dict)
|
|
"""Extra environment for the claude subprocess (always passed through)."""
|
|
|
|
env_keep: tuple[str, ...] = ()
|
|
"""Extra inherited variable names to let through the env whitelist."""
|
|
|
|
max_turns: int | None = None
|
|
max_buffer_size: int = 16 * 1024 * 1024
|
|
session_store_flush: str = "eager"
|
|
"""``eager`` mirrors every transcript frame as it lands, so a gateway
|
|
killed mid-turn loses at most the frame in flight."""
|
|
|
|
|
|
class Prompts(BaseModel):
|
|
"""Prompt assembly per conversation kind: the granules, in order.
|
|
|
|
A kind left ``None`` is not served; ``ClaudeAgent.kinds`` follows from
|
|
the kinds set here.
|
|
"""
|
|
|
|
model_config = ConfigDict(frozen=True)
|
|
|
|
master: tuple[PromptSource, ...] | None = None
|
|
branch: tuple[PromptSource, ...] | None = None
|
|
deep: tuple[PromptSource, ...] | None = None
|
|
job: tuple[PromptSource, ...] | None = None
|
|
fork: tuple[PromptSource, ...] | None = None
|
|
|
|
def for_kind(self, kind: Kind) -> tuple[PromptSource, ...] | None:
|
|
return getattr(self, kind)
|
|
|
|
@property
|
|
def kinds(self) -> tuple[Kind, ...]:
|
|
return tuple(k for k in KINDS if self.for_kind(k) is not None)
|
|
|
|
|
|
class SkillSets(BaseModel):
|
|
"""Skill-set directories per conversation kind; ``None`` means none."""
|
|
|
|
model_config = ConfigDict(frozen=True)
|
|
|
|
master: tuple[Path, ...] | None = None
|
|
branch: tuple[Path, ...] | None = None
|
|
deep: tuple[Path, ...] | None = None
|
|
job: tuple[Path, ...] | None = None
|
|
fork: tuple[Path, ...] | None = None
|
|
|
|
def for_kind(self, kind: Kind) -> tuple[Path, ...]:
|
|
return getattr(self, kind) or ()
|
|
|
|
|
|
class ClaudeAgent(BaseAgent):
|
|
cwd: Path
|
|
system_prompt: str = ""
|
|
"""Verbatim prompt for agents without ``prompts`` (tests, one-offs)."""
|
|
|
|
prompts: Prompts = Field(default_factory=Prompts)
|
|
kinds: tuple[Kind, ...] = ()
|
|
"""Conversation kinds this agent serves; ``create``/``spawn`` reject the rest.
|
|
Defaults to what ``prompts`` covers, or ``("deep",)`` for a verbatim prompt."""
|
|
|
|
skill_sets: tuple[Path, ...] | SkillSets = ()
|
|
"""Skill-set directories, each a local plugin: one tuple for every kind,
|
|
or a ``SkillSets`` giving each kind its own (``None`` = no skills)."""
|
|
|
|
gateway_tools: tuple[str, ...] = ()
|
|
"""Gateway tools exposed in-process (``read_conversation``, ``spawn``,
|
|
``say``, ``schedule``, ``inject``); empty = no gateway MCP server."""
|
|
|
|
options: ClaudeOptions = Field(default_factory=ClaudeOptions)
|
|
policy: tuple[PolicyRule, ...] = ()
|
|
"""``PreToolUse`` rules, run in order on every tool call; the first
|
|
``Deny`` is what the model reads back. See ``agents/policy.py``."""
|
|
|
|
@model_validator(mode="after")
|
|
def _kinds_follow_prompts(self) -> ClaudeAgent:
|
|
covered = self.prompts.kinds
|
|
if not self.kinds:
|
|
object.__setattr__(self, "kinds", covered or ("deep",))
|
|
elif covered:
|
|
missing = [k for k in self.kinds if k not in covered]
|
|
if missing:
|
|
msg = f"agent {self.name!r} serves {missing} without a prompt"
|
|
raise ValueError(msg)
|
|
return self
|
|
|
|
@property
|
|
def thinking_is_progress(self) -> bool:
|
|
"""A progress model, unless ``display: summarized`` brings reasoning."""
|
|
summarized = (self.options.thinking or {}).get("display") == "summarized"
|
|
return self.model in PROGRESS_MODELS and not summarized
|
|
|
|
def prompt_for(self, kind: Kind) -> tuple[PromptSource, ...] | None:
|
|
return self.prompts.for_kind(kind)
|
|
|
|
def skills_for(self, kind: Kind) -> tuple[Path, ...]:
|
|
if isinstance(self.skill_sets, SkillSets):
|
|
return self.skill_sets.for_kind(kind)
|
|
return self.skill_sets
|
|
|
|
def serves(self, kind: str) -> bool:
|
|
return kind in self.kinds
|