Designing a Production Agent Harness

The canonical ReAct loop is 15 lines: prompt the model, parse Thought/Action/Observation, call the tool, repeat. It works in demos. It fails in production because it assumes:

• The model always returns valid JSON for tool calls.
• Tools never time out, return errors, or exceed rate limits.
• The conversation can grow forever without hitting the context window.
• A crash loses no state — you can just restart.
• There is no way to cancel a runaway agent mid-flight.

None of those assumptions hold at scale. A production harness is what you build to make each assumption explicitly false: schema validation gates every tool call, retries handle transient errors, a token budget caps context growth, an abort channel stops the loop from the outside, and a journal lets you resume after a crash. Each of those mechanisms is a seam where bugs hide — and where most teams skip the work.

Each turn of the loop does exactly four things in order:

1. Assemble context — reconstruct the messages array from the journal (not from in-memory state, which evaporates on crash).
2. Call the model — pass a max_tokens cap and (if the API supports it) a system-level token budget header.
3. Dispatch — if the model returns a tool call, run it; if it returns a final answer, yield it to the caller and stop.
4. Record — write every LLM response and every tool result to the journal before moving on.

The loop must be a while True with an explicit break condition — not recursion. Recursion blows the call stack for long agent runs and makes it much harder to inject an abort check. Keep the loop body a single function so you can unit-test each step in isolation.

A registry maps a string name to a callable and its JSON Schema descriptor. Register tools at startup, not inline.

from dataclasses import dataclass, field
from typing import Callable, Any

@dataclass
class ToolDef:
    name: str
    description: str
    parameters: dict          # JSON Schema object
    fn: Callable[..., Any]
    timeout_s: float = 30.0
    max_retries: int = 3

class ToolRegistry:
    def __init__(self):
        self._tools: dict[str, ToolDef] = {}

    def register(self, tool: ToolDef) -> None:
        if tool.name in self._tools:
            raise ValueError(f"Duplicate tool: {tool.name}")
        self._tools[tool.name] = tool

    def get(self, name: str) -> ToolDef:
        if name not in self._tools:
            raise KeyError(f"Unknown tool: {name!r}")
        return self._tools[name]

    def as_api_tools(self) -> list[dict]:
        """Return the list the LLM API expects."""
        return [
            {"name": t.name, "description": t.description,
             "input_schema": t.parameters}
            for t in self._tools.values()
        ]

The as_api_tools() method means the schema lives in exactly one place — no drift between what the model sees and what you validate against.

Never trust the model's arguments blindly. Validate against the same JSON Schema you sent to the API, then coerce types if needed.

import jsonschema

def validate_args(tool: ToolDef, raw_args: dict) -> dict:
    try:
        jsonschema.validate(instance=raw_args, schema=tool.parameters)
    except jsonschema.ValidationError as exc:
        # Return an error observation instead of raising — the model
        # can self-correct on the next turn.
        raise ToolValidationError(
            f"Invalid args for {tool.name!r}: {exc.message}"
        ) from exc
    return raw_args

When validation fails, feed the error back as an observation rather than crashing the loop. The model usually self-corrects within one turn. If it fails N times in a row on the same tool call, that is a hard abort — the model is stuck.

Heads up: JSON Schema additionalProperties: false is your friend. Without it, a model can hallucinate extra fields and the schema passes anyway.

Not all tool errors are equal. Classify them before deciding whether to retry:

import asyncio, httpx

async def call_with_retry(tool: ToolDef, args: dict) -> str:
    delay = 1.0
    for attempt in range(tool.max_retries + 1):
        try:
            return await asyncio.wait_for(
                tool.fn(**args), timeout=tool.timeout_s
            )
        except asyncio.TimeoutError:
            obs = f"Tool timed out after {tool.timeout_s}s"
        except httpx.HTTPStatusError as e:
            if e.response.status_code in (401, 403):
                raise  # fatal
            obs = f"HTTP {e.response.status_code}: {e.response.text[:200]}"
        if attempt < tool.max_retries:
            await asyncio.sleep(delay)
            delay = min(delay * 2, 60)
    return obs  # last error becomes the observation

The key insight: retry at the tool execution layer, not at the LLM layer. Retrying the LLM after a tool error wastes tokens and context.

Context windows are large but not infinite, and large contexts are slow and expensive. A token budget enforces two limits:

Input budget — the maximum tokens you'll send to the model per turn. When the journal grows past this, you need a compaction strategy (summarize older turns, drop raw tool outputs, keep only the last N observations).

Output budget — set max_tokens on every API call. Never omit it; some models will happily emit 8k tokens when a tool call needs 50.

TOKEN_BUDGET = 80_000   # leave 48k headroom in a 128k window

def trim_messages(messages: list[dict], tokenizer) -> list[dict]:
    total = sum(len(tokenizer.encode(m["content"])) for m in messages)
    while total > TOKEN_BUDGET and len(messages) > 2:
        # Drop oldest tool result first (they're the least load-bearing)
        for i, m in enumerate(messages):
            if m["role"] == "tool":
                removed = messages.pop(i)
                total -= len(tokenizer.encode(removed["content"]))
                break
    return messages

Anthropic's API also supports a budget_tokens field in the thinking block — use it for reasoning models to cap chain-of-thought spend separately from the answer.

A runaway agent can burn through API budget or take destructive actions if left unchecked. You need a way to stop it from outside the loop — from a UI cancel button, a deadline timer, or a watchdog process.

The cleanest pattern is an asyncio.Event (or a threading.Event) checked at the top of every iteration:

async def run_agent(
    task: str,
    registry: ToolRegistry,
    journal: Journal,
    abort: asyncio.Event,
) -> str:
    while not abort.is_set():
        messages = journal.to_messages()
        response = await llm_call(messages, registry.as_api_tools())
        journal.append_llm(response)

        if response.stop_reason == "end_turn":
            return response.content[-1].text

        for tool_use in response.tool_use_blocks():
            if abort.is_set():
                journal.append_tool(tool_use.id, "[aborted by caller]")
                return "[agent aborted]"
            result = await call_with_retry(
                registry.get(tool_use.name), tool_use.input
            )
            journal.append_tool(tool_use.id, result)

    return "[agent aborted]"

Never check the abort signal only at the top of the outer loop — check it before each individual tool call too, since a single tool can block for seconds.

If the process dies mid-run, all in-memory state is gone. A journal persists every event to disk (or a database) so you can resume — or at least reconstruct what happened.

import json, pathlib
from datetime import datetime, timezone

class Journal:
    def __init__(self, path: pathlib.Path):
        self.path = path
        self._entries: list[dict] = []
        if path.exists():
            self._entries = json.loads(path.read_text())

    def _flush(self):
        self.path.write_text(json.dumps(self._entries, indent=2))

    def append_llm(self, response) -> None:
        self._entries.append({"ts": _now(), "role": "assistant",
                               "raw": response.model_dump()})
        self._flush()

    def append_tool(self, tool_use_id: str, result: str) -> None:
        self._entries.append({"ts": _now(), "role": "tool",
                               "tool_use_id": tool_use_id, "content": result})
        self._flush()

    def to_messages(self) -> list[dict]:
        """Reconstruct the messages list the LLM API expects."""
        ...

def _now() -> str:
    return datetime.now(timezone.utc).isoformat()

Flush after every write — not in bulk. A crash between two writes is survivable; losing a batch is not. On high-throughput systems, replace the JSON file with an append-only SQLite table using WAL mode.

A harness without observability is a black box. At minimum, emit structured log events (JSON lines to stdout works fine) for:

• Turn start — run_id, turn, input_tokens (estimated)
• Tool dispatch — tool_name, args (redact secrets), attempt
• Tool result — tool_name, latency_ms, status (ok / error / timeout)
• Abort — reason, turn count at abort time
• Final answer — run_id, total_turns, total_tokens, wall_time_ms

Avoid logging full LLM responses inline — they're huge and often contain PII. Log a response_id and store the raw response separately.

Heads up: if you use OpenTelemetry, wrap each turn as a span with run_id as the trace root. This gives you waterfall latency breakdowns across multi-tool turns for free.

The gap between a toy loop and a production harness isn't a single feature — it's the accumulation of seams, each one a failure mode you've closed:

• Registry closes the "tool not found" and schema-drift failure modes.
• Validation closes the "model passed garbage args" failure mode.
• Retry + classify closes the "transient error kills the run" failure mode.
• Token budget closes the "context overflow kills the run" failure mode.
• Abort signal closes the "runaway agent burns budget / takes bad action" failure mode.
• Journal closes the "crash loses all progress" failure mode.

You do not need all of these on day one. Start with the journal (cheapest, highest leverage) and the schema validation (saves hours of debugging malformed tool calls). Add retries next. Token budgets and abort signals become urgent once your agents run tasks longer than 30 seconds.

The pattern generalizes across APIs — Anthropic tool use, OpenAI function calling, Google Gemini function declarations all share the same loop shape. The registry and journal are model-agnostic.