agent-loop

agent-loop

Production patterns for building autonomous agent loops with the Anthropic API. Covers the Session/Harness/Registry/Tool abstraction, safe development mode via DRY_RUN, APScheduler cron integration, and observability hooks. Designed for systems that run unsupervised on a schedule and must never silently bill or silently fail.

Core Abstraction

Four objects, one responsibility each:

Session   — append-only event log (what happened)
Harness   — drives the Claude API loop (who decides)
Registry  — tool catalogue (what can be called)
Tool      — unit of action (what gets done)

# RunContext carries credentials and settings — never sent to Claude
@dataclass
class RunContext:
    anthropic_api_key: str
    settings: Settings
    system_code: str        # e.g. "LI-01", "OB-03"
    client_id: str
    run_id: str = field(default_factory=lambda: str(uuid4()))

DRY_RUN Guard — First Line of Safety

Always gate API calls. A scheduler firing 39 jobs at noon will call the Anthropic API 39 times if this is missing.

# settings.py
class Settings(BaseSettings):
    # Default TRUE — scheduler fires, tools run, but NO Claude API calls.
    # Flip to false only when ready to go live.
    dry_run: bool = True

# harness.py — guard before the loop
async def run(self, system_prompt: str, initial_message: str, max_turns: int = 10) -> str:
    if self.ctx.settings.dry_run:
        logger.info("DRY_RUN=true — skipping Claude API call for %s", self.ctx.system_code)
        return "[dry_run] No API call made — set DRY_RUN=false to enable"

    # ... API loop follows

# .env — explicit is better than implicit
DRY_RUN=true    # change to false only when going live

Harness — The Agent Loop

class Harness:
    def __init__(self, ctx, session, registry, model=None, max_tokens=1024):
        self.ctx      = ctx
        self.session  = session
        self.registry = registry
        self.model    = model or ctx.settings.anthropic_model
        self._client  = anthropic.AsyncAnthropic(api_key=ctx.anthropic_api_key)

    async def run(self, system_prompt, initial_message, max_turns=10) -> str:
        messages = [{"role": "user", "content": initial_message}]
        tools    = self.registry.to_anthropic()

        if self.ctx.settings.dry_run:
            return "[dry_run] No API call made"

        for turn in range(max_turns):
            self.session.record_claude_request(
                messages=messages,
                system_prompt_hash=hashlib.sha256(system_prompt.encode()).hexdigest()[:16],
                model=self.model,
                max_tokens=self.max_tokens,
            )

            try:
                response = await self._client.messages.create(
                    model=self.model, max_tokens=self.max_tokens,
                    system=system_prompt, tools=tools, messages=messages,
                )
            except anthropic.APIError as exc:
                self.session.error("anthropic_api_error", str(exc))
                raise DeadLetterError(f"Anthropic API error: {exc}") from exc

            self.session.record_claude_response(
                stop_reason=response.stop_reason,
                input_tokens=response.usage.input_tokens,
                output_tokens=response.usage.output_tokens,
                tool_calls=[b.name for b in response.content if b.type == "tool_use"],
            )

            messages.append({"role": "assistant", "content": response.content})

            if response.stop_reason == "end_turn":
                return "\n".join(b.text for b in response.content if hasattr(b, "text"))

            if response.stop_reason != "tool_use":
                raise DeadLetterError(f"Unexpected stop_reason: {response.stop_reason}")

            # Execute all tool calls
            tool_results = []
            for block in response.content:
                if block.type != "tool_use":
                    continue
                tool   = self.registry.get(block.name)
                result = await self._execute_tool(tool, block)
                tool_results.append(result.to_anthropic())

            messages.append({"role": "user", "content": tool_results})

        raise DeadLetterError(f"Reached max_turns={max_turns} without end_turn")