Harness Patterns

Run the agent as a loop: build the request, call the model, run any tools it asks for, and repeat until it stops asking.

Every harness has a loop at its centre, wrapped around the call to the model. Each pass sends the system prompt (the standing instructions) and the transcript (the conversation so far). It also sends the tool schemas (descriptions of the tools the model may use). Then the harness looks at the reply. If the reply has tool_use blocks (the model's requests to run a tool), the harness runs those tools. It puts the matching tool_result blocks (the tools' answers) into the next user message and goes round again. If stop_reason (the field that says why the model stopped) is end_turn, the model has finished. The loop ends and the final text goes to the user. Other stop reasons need their own handling. max_tokens means the reply was cut off at the length limit. A cut-off tool_use must never be run. pause_turn means a server-side tool (a tool the provider runs) paused a long turn. The harness should then send the request again to continue. The harness checks its guards at the start of every pass. These are max_turns (a limit on turns), a token budget (a limit on how much text the run may use) and a cancel flag. This is also where it handles compaction (shrinking old context to save space) and user messages waiting in a queue.

while turn < max_turns and not cancelled:
    turn += 1
    resp = model.create(system=sys_prompt,
                        messages=msgs,
                        tools=tools)
    msgs.append(resp.as_message())
    if resp.stop_reason == 'max_tokens':
        raise Truncated(resp)  # never run it
    if resp.stop_reason == 'pause_turn':
        continue  # resend to resume
    calls = resp.tool_use_blocks()
    if not calls:
        break  # end_turn: answer is ready
    msgs.append(run_tools(calls))

Be carefulA loop with no turn limit or budget check can run forever. This happens when the model keeps calling the same failing tool.

Give each part of the harness its own update hook, and let the loop call every hook once per turn.

Between model calls, a harness has many small jobs that must move forward a little each turn. The token meter works out how much text has been used so far. The compactor watches the context window (the text the model can see at one time). It checks whether the window is nearly full and must be shrunk. A todo tracker reminds the model if its task list is out of date. A poller collects background tasks that have finished. The harness does not write all of this into the loop itself. Instead, each subsystem (each separate part of the harness) registers an object with an on_turn(ctx) method. Here ctx is the context object that holds this turn's information. The loop goes through the list of registered objects once per turn. So adding a new behaviour means one new registration, not a change to the loop.

class Compactor:
    def on_turn(self, ctx):
        if ctx.tokens > 0.8 * ctx.window:
            ctx.msgs = summarise(ctx.msgs)

systems = [BudgetMeter(), Compactor(),
           TaskPoller(), TodoNudger()]

for turn in agent_loop():
    for s in systems:
        s.on_turn(turn.ctx)

Be carefulSometimes one update needs another update's result. Then the order of registration matters, so write that order down or merge those parts.

Make each phase of the loop its own state object, so each state decides how to react to messages, interrupts and tool results.

An agent session moves through clear phases: Idle, CallingModel, AwaitingApproval, RunningTool and Done. The same event means different things in each phase. A new user message starts a turn when the session is Idle. The same message waits in a queue while a tool is running (RunningTool). Pressing Esc in CallingModel stops the reply that is streaming in. Pressing Esc in AwaitingApproval counts as saying no. Each phase can be written as a state class with methods such as on_message, on_interrupt and on_tool_result. This replaces a tangle of if-statements. It also makes illegal transitions impossible to write by accident. A transition is a move from one state to another. An illegal one is, for example, running a tool without approval.

class AwaitingApproval(State):
    def on_approve(self, s):
        return RunningTool(self.call)
    def on_deny(self, s):
        s.add_result(self.call, 'denied')
        return CallingModel()
    def on_message(self, s, m):
        s.queue.append(m)  # hold it
        return self

Be carefulDo not let tools or hooks set the state field directly. Every change of state should go through the current state's methods.

Fix the order of a turn's steps in a base class, and let each agent type change only the hooks it needs.

A harness base class (shared parent code that other classes build on) defines run_turn(). This method builds the system prompt, trims the context, calls the model, sends out tool calls and records the turn. The method is final. That means subclasses (classes built on the base class) cannot replace it. A coding agent overrides (replaces) system_prompt() to add the repository's rules. It also overrides on_tool_result() to run a linter (a code style checker) after edits. A research subagent overrides only select_tools(), so it can use just search and fetch. The outline never changes. So certain rules hold for every agent type. For example, every turn is logged to the transcript, and the token budget is checked before each call.

class BaseAgent:
    def run_turn(self, msgs):  # fixed
        sp = self.system_prompt()  # hook
        resp = self.model.create(
            system=sp, messages=msgs)
        for call in resp.tool_calls:
            out = self.tools.run(call)
            self.on_tool_result(call, out)
        self.transcript.record(resp)
class CodingAgent(BaseAgent):
    def system_prompt(self):
        return BASE + REPO_RULES

Be carefulIf subclasses start to override run_turn() itself, the guarantees are lost. Mark it final and add a new hook instead.

Pass each tool call through an ordered chain of hooks that can block, change or pass it on. Only the chain's end lets it run.

Before a tool call runs, the harness passes it through a chain of PreToolUse hooks. A hook is a small piece of code that runs at a set moment, here just before a tool is used. The chain might start with an allowlist check (a list of allowed tools). Next comes a path guard that blocks writes outside the workspace (the project folder the agent works in). Then a secrets scanner looks for passwords and keys. Last comes a prompt that asks the user to approve. Each hook gets the call and returns deny, modify or pass. A deny stops the chain at once. It becomes an error tool_result (the tool's answer, marked as an error) that the model can read. A pass hands the call to the next hook. Hooks managed by the organisation sit at the front of the chain. User settings cannot remove or override them. So a user hook that says allow can only skip the final approval prompt, never a required guard. Users and plugins add their own hooks after the managed ones, without changing harness code.

class PathGuard(Hook):
    def handle(self, call, nxt):
        p = call.input.get('path')
        if p:
            rp = Path(ROOT, p).resolve()
            if not rp.is_relative_to(ROOT):
                return Deny('outside root')
        return nxt(call)

ROOT = Path(WORKSPACE).resolve()
chain = [Allowlist(), PathGuard(),
         SecretScan(), AskUser()]
result = run_chain(chain, call, execute)

Be carefulIn a long chain, it is hard to see which hook made the decision. Log the name of the deciding hook with every allow and every deny.

Read the model's structured output as action objects from a small grammar (a set of language rules), then run them on the workspace.

Some harnesses let the model answer in a compact action language instead of one tool call per step. Examples are an edit block format, or a plan DSL with steps like edit, run and if_fail. A DSL (domain-specific language) is a small language made for one job. The harness parses (reads and breaks down) that text into a tree of expression nodes: Seq, Edit, Run and OnFail. Each node has an interpret(ctx) method that does its part of the work on the workspace (the project files). It goes through the same permission checks and sandbox as an ordinary tool call. A sandbox is a locked-down area where code runs safely. Sometimes the program arrives inside a tool_use input (the model's request to run a tool). Then parse errors go back as an error tool_result (the tool's answer, marked as an error), so the model can fix its syntax. Other times the program arrives as plain text from the model. Then errors go back in the next user message, because a tool_result needs a matching tool_use id.

class Seq(Expr):
    def interpret(self, ctx):
        return [s.interpret(ctx)
                for s in self.steps]

class Edit(Expr):
    def interpret(self, ctx):
        return ctx.fs.patch(self.path,
                            self.diff)

tree = parse(model_text)  # Seq([Edit, Run])
results = tree.interpret(ctx)

Be carefulKeep the grammar tiny. Once it needs variables and loops, you are building a programming language, and a real language in a sandbox is safer.

Save a snapshot of the session's messages and file edits before each risky step, so the user can go back to any earlier turn.

Before each turn that edits files, the session makes a checkpoint. A checkpoint is a copy of the message list plus the original contents of every file the turn is about to change. A checkpoint store keeps these snapshots by turn number, without looking inside them. When the user runs a rewind command, the session restores the chosen checkpoint. It rewrites the files and cuts the transcript back to that turn. So both the conversation and the workspace (the project files) return to that point. Long sessions keep only recent checkpoints, or store file diffs (only the changes) to save disk space.

class Session:
    def checkpoint(self):
        files = self.fs.copy(self.touched)
        return Snapshot(self.msgs[:], files)
    def restore(self, snap):
        self.msgs = list(snap.msgs)
        self.fs.write_all(snap.files)

store[turn] = session.checkpoint()
session.restore(store[6])  # /rewind 6

Be carefulSnapshots cannot undo actions outside the workspace, such as a sent message or a deployed build. Tell the user this when they rewind.

Go through stream events, turns or stored messages one at a time with next(), without loading or exposing the whole source.

A streaming model reply arrives as server-sent events (SSE: small messages the server pushes one after another). Their names include message_start, content_block_delta and content_block_stop. The SDK (the provider's code library) wraps the raw stream of bytes in an async iterator. An async iterator hands out items one at a time as they arrive, without freezing the program. So the harness can write async for event in stream and show text deltas (new pieces of text) as they come in. The same idea is used in other places. A transcript iterator reads JSONL lines (one JSON record per line) from disk only when needed. A paginated iterator over a long history fetches the next page only when the consumer asks. Consumers never care whether the items come from a network socket, a file or an API.

async for ev in client.stream(req):
    if (ev.type == 'content_block_delta'
        and ev.delta.type == 'text_delta'):
        ui.append(ev.delta.text)
    elif ev.type == 'message_stop':
        break

def turns(path):  # lazy over a JSONL file
    with open(path) as f:
        for line in f:
            yield json.loads(line)

Be carefulAn iterator over a live stream can be read only once. If both the screen and the transcript writer need the events, keep a copy of them.

Treat a message, a tool result and a single text block through the same interface, because content blocks sit inside each other.

A message's content is a list of blocks: text, image, tool_use and tool_result. A tool_result (the answer from a tool) can itself contain text and image blocks. The harness gives every block the same small interface (the same set of methods), such as token_count(), render() and redact(). Leaves (blocks with nothing inside them) answer directly. Containers like Message and ToolResult ask their children and combine the answers. So counting the tokens (the small text units models count) of a whole transcript becomes one call on the root (the top of the tree). It no longer needs nested loops that check each block's type.

class TextBlock(Block):
    def tokens(self):
        return count(self.text)

class ToolResult(Block):
    def tokens(self):
        return 8 + sum(c.tokens()
                       for c in self.content)

total = sum(m.tokens() for m in transcript)

Be carefulDo not force methods onto the shared interface when they make no sense for every block. A leaf with fifteen methods that do nothing is a warning sign.

Keep content block classes stable, and put each operation that runs across them, like token counting or redaction, in its own visitor class.

Content blocks (the pieces of a message, such as text, images and tool calls) rarely change. But the list of operations run over them keeps growing. The harness counts tokens, redacts (hides) secrets before logging, and renders blocks to the terminal or to HTML. It also converts them to another provider's format. With Visitor, each block has a single accept(v) method. That method calls the matching method on the visitor v, such as v.visit_text(self) or v.visit_tool_use(self). A new operation is a new visitor class with one method per block type. The block classes and the code that walks the transcript stay the same.

class ToolUseBlock(Block):
    def accept(self, v):
        return v.visit_tool_use(self)

class Redactor(Visitor):
    def visit_text(self, b):
        return scrub(b.text)
    def visit_tool_use(self, b):
        return scrub_json(b.input)  # deep

clean = [b.accept(Redactor()) for b in blocks]

Be carefulAdding a new block type means updating every visitor. If new block types arrive often, use Acyclic Visitor or plain methods on the blocks instead.

Let each visitor handle only the block types it knows, so a new content block type never forces edits elsewhere.

New content block types keep arriving: thinking blocks, citations, documents and server tool results. In a classic Visitor, every new type means adding a new method to every visitor. In the acyclic version, ImageBlock.accept(v) checks whether v implements ImageVisitor. ImageVisitor is a small interface, meaning a promised set of methods, for image blocks. The block calls visit_image only if v has it. A thumbnail generator implements only ImageVisitor. A new DocumentBlock comes with its own DocumentVisitor interface, and the thumbnail generator does not change. Safety visitors are the exception. A secret redactor must treat any block type it does not handle as an error. So a new DocumentBlock is blocked until the redactor learns to scan it.

@runtime_checkable
class ImageVisitor(Protocol):
    def visit_image(self, b): ...

class ImageBlock(Block):
    def accept(self, v):
        if isinstance(v, ImageVisitor):
            return v.visit_image(self)
        return v.unhandled(self)  # skip/raise

class Thumbnailer(ImageVisitor):
    def visit_image(self, b):
        return shrink(b.data, 256)

Be carefulSkipping unknown types is safe for a thumbnail maker. It is not safe for a redactor or a DLP scanner (a tool that stops sensitive data from leaking). These must fail closed and reject any block type they do not handle.

Choose the handler from two runtime types at once: the incoming event's type and the session's current state.

The harness receives several event types: UserMessage, Interrupt, ToolResult and ApprovalReply. Meanwhile the session can be in several states: Idle, Streaming and RunningTool. The right action depends on both. A UserMessage in Idle starts a turn, but in RunningTool it waits in a queue. An Interrupt in Streaming cancels the request, but in RunningTool it kills the child process (the separate program running the tool). The event's dispatch(state) method calls state.on_user_message(self) or state.on_interrupt(self). So the language's own method lookup picks the right cell in the event-by-state grid. No type switches (long if-else checks on type) are needed.

class Interrupt(Event):
    def dispatch(self, state):
        return state.on_interrupt(self)

class RunningTool(State):
    def on_interrupt(self, ev):
        self.proc.kill()
        return Idle()
    def on_user_message(self, ev):
        self.pending.append(ev)
        return self

Be carefulThe grid grows as types times states. When most cells share a default, a small lookup table is easier to read than many methods.

Pass functions into the loop, and the loop calls them when a token streams in, a tool finishes or a turn ends.

Several clients share one agent loop: a terminal UI, a web client, an SDK script and an eval runner. An eval runner is a tool that tests the agent automatically. Each one wants to react at different moments. The loop accepts callbacks (functions it will call later) such as on_text_delta, on_tool_start, on_tool_end and on_turn_complete. It calls each one at its moment. The terminal prints the new pieces of text (deltas). The web client pushes them over a websocket, a live two-way connection to the browser. The eval runner ignores deltas but records every tool call. The loop itself never imports any UI code.

def show(delta):
    term.write(delta.text)

def save(msg):
    transcript.append(msg)

agent.run(prompt,
          on_text_delta=show,
          on_tool_end=lambda c, r: log(c, r),
          on_done=save)

Be carefulA slow or failing callback can block or crash the loop. Give callbacks a time limit and catch their errors, so the turn still finishes.

Bundle everything one turn needs (ids, budget, working folder, permissions, cancel signal) into one object passed to every component.

Every turn touches many components: the model client, the tool runner, hooks, the tracer and the code that starts subagents. Each one needs some of these facts: the session id, turn number, working directory, permission mode, remaining token budget, abort signal and trace span. A trace span is one timed record of work, used for tracing. The harness builds a TurnContext at the start of the turn. It passes that single object down every call. Say you add a field, such as the active MCP servers. MCP is a standard way to connect tools to an AI model. You change only the context class and the code that reads it. You do not change every function signature in between, meaning the list of inputs each function takes.

@dataclass
class TurnContext:
    session_id: str
    turn: int
    cwd: Path
    perms: PermissionMode
    budget: TokenBudget
    cancel: AbortSignal

ctx = TurnContext(sid, n, cwd, mode, b, sig)
resp = model.call(msgs, ctx)
out = tools.run(resp.tool_calls, ctx)

Be carefulA context object that collects every global value in the system becomes a hidden bag of dependencies. Keep it for one turn only, and keep it small.

Pass one collector object through a series of steps, so each step adds its findings instead of returning separate results to merge.

A code review subagent (a helper agent started by the main one) runs several checks. It uses grep (a text search tool) to find TODO notes and scans the diff (the code changes) for secrets. It also runs the tests and checks the changelog. Each check gets the same Findings object. It adds issues with the file, line and severity. After the last check, the subagent turns the collected findings into one report message for the orchestrator (the agent in charge). The same shape appears when a harness adds up usage across smaller calls. One UsageCollector is passed into every model call and tool call. The turn total is read from it at the end.

findings = Findings()
check_todos(diff, findings)
scan_secrets(diff, findings)
run_tests(repo, findings)

def scan_secrets(diff, out):
    for ln in diff.added_lines():
        if KEY_RE.search(ln.text):
            out.add(ln.path, ln.no, 'high')
report = findings.to_markdown()

Be carefulIf steps run in parallel, the collector needs a lock, so only one step writes at a time. Otherwise each step should return its own list to merge afterwards.

Prepare context in fixed stages: find candidates, rerank them by relevance, shorten the best ones, then insert them into the prompt.

Before calling the model, a retrieval-augmented harness runs a context pipeline. Retrieval-augmented means it looks up documents and adds them to the prompt. The retriever pulls fifty chunks (short pieces of text) from a vector store and a keyword index. A vector store finds text by meaning, and a keyword index finds it by words. A reranker scores the chunks against the query and keeps eight. A compressor trims each one to the sentences that matter. An injector formats them as document blocks in the prompt, with source ids. Each stage takes the query and a list of chunks, and returns a new list. So a team can swap the reranker model without touching the other stages. They can also add a stage, such as one that removes duplicates or one that checks access rights. Evals (automated quality tests) can score each stage on its own.

stages = [retrieve(k=50), rerank(top=8),
          compress(max_tokens=2000), inject]

def run(query):
    data = []
    for stage in stages:
        data = stage(query, data)
    return data  # prompt-ready blocks

prompt = run(user_msg)

Be carefulAny stage can drop the one chunk that mattered. Log what each stage removed, so a bad answer can be traced back to its stage.

Check a set of written rules against each turn's facts to decide routing, approvals and compaction, without hard-coded branches.

Each turn, the harness gathers facts. These include context usage, the requested tool and a risk class for its arguments. They also include the user's plan tier, the working directory and the time of day. A rule engine checks rules loaded from a settings file. One rule says: if the context is over 80 percent full, compact it (shrink it) first. Another says: if a command classifier marks the shell command as destructive, require approval. A command classifier is code that reads a command and labels how risky it is. A third says: if the task is classified as simple, send it to a smaller model. Matching rules produce actions, and the loop applies them in priority order. Operators change behaviour by editing rules as data. The rules can be tested against recorded turns.

RULES = [  # loaded from rules.yaml
  {'if': {'ctx_pct': {'gt': 80}},
   'then': 'compact_first', 'prio': 1},
  {'if': {'cmd_risk': {'eq': 'high'}},
   'then': 'require_approval', 'prio': 0},
  {'if': {'task': {'eq': 'simple'}},
   'then': 'use_small_model', 'prio': 2}]

hits = [r for r in RULES
        if match(r['if'], facts)]
hits.sort(key=lambda r: r['prio'])
for r in hits:
    apply(r['then'])

Be carefulRules that affect each other are hard to reason about, for example when one rule's action changes another rule's facts. Keep actions independent, or check the rules only once per turn.

Run deep recursive planning as a flat loop, where each step returns the next step instead of calling it.

Some agents refine a plan one subgoal at a time. This is naturally written as a tail-recursive function, where the function's last act is to call itself. Here, plan(goal, steps) asks the model for the next subgoal and calls itself with the longer list. Python has no tail-call elimination, so it does not reuse memory for such calls. A long plan driven by the model keeps every in-between frame (a record of each unfinished call). It can then hit Python's recursion limit. With a trampoline, plan returns either Bounce(fn, args) for the next step or Done(result). A small driver loop keeps calling the bounces. This also gives the harness a natural place between steps to check the turn budget, save a checkpoint and respect a cancel request. A trampoline only flattens tail calls. A planner that branches into a tree of subgoals needs an explicit work stack instead, which is a list of pending tasks it manages itself.

def plan(goal, acc):
    sub = model.next_subgoal(goal, acc)
    if sub is None:
        return Done(acc)
    return Bounce(plan, goal, acc + [sub])

def trampoline(r):
    while isinstance(r, Bounce):
        budget.check(); cancel.check()
        r = r.fn(*r.args)
    return r.value

Be carefulTrampolining adds an extra layer to code that rarely goes deep. Use it only when the depth of recursion depends on model output you do not control.

Turn every tool call into an object that the harness can queue, log, approve, retry or undo.

A tool call is the model's request to run a tool, like reading a file. The model sends it as a tool_use block (a part of its reply that names the tool, its inputs and a call id). The harness wraps that block in a ToolCommand object. The object holds the tool name, the arguments and the call id. It then travels through the permission gate (the step that decides if the call may run), the execution queue and the transcript (the saved record of the conversation). The call is now data, not a function that runs at once. So the harness can pause it for the user to approve. It can replay it in an eval (an automated test of the agent). It can log it for audit. And it can run a compensating undo (an action that reverses the effect) when a later step fails.

class ToolCommand:
    def __init__(self, tool, args, cid):
        self.tool, self.args = tool, args
        self.id = cid
    def execute(self):
        return self.tool.run(**self.args)
    def undo(self):
        return self.tool.revert(self.id)

cmd = ToolCommand(tools['write_file'],
                  block.input, block.id)
queue.put(cmd)

Be carefulSome commands change the outside world, like sending an email or paying an invoice. These often cannot be undone. Mark them as irreversible and ask for approval before they run.

Keep one tool name and schema, but swap the code behind it, such as which backend a search tool uses.

The model sees one search tool. Its input schema (the fixed list of inputs it accepts) is query and max_results. Its description does not promise any particular backend (the system that does the real searching). Behind it, the harness holds a SearchStrategy interface with several versions. One uses a hosted web search API. One uses a vector store (a database for search by meaning) over the user's documents. One uses local ripgrep (a fast file search program) when the machine is offline. The config or the workspace type picks the strategy when the session starts. So the tool definition in the request stays byte-identical, exactly the same down to each character. That keeps the prompt cache hitting. The prompt cache is saved work the provider can reuse when a request starts the same way. Evals (automated tests of the agent) plug in a canned strategy that returns fixed results.

class SearchStrategy(Protocol):
    def search(self, q, k): ...

class SearchTool:
    name = 'search'
    def __init__(self, strategy):
        self.strategy = strategy
    def run(self, query, max_results=5):
        hits = self.strategy.search(
            query, max_results)
        return format_hits(hits)

tool = SearchTool(STRATEGIES[cfg.search])

Be carefulStrategies may return results of different shape or quality. Then the model cannot learn what the tool does. Convert every strategy's output to one format.

Wrap MCP servers and outside APIs so each one looks like a native tool, with the harness's own name, schema and run method.

The agent loop (the cycle of calling the model and running tools) knows one contract. A Tool has a name, a description, an input_schema, and run(args) that returns a ToolResult. MCP is a standard way to connect tools to an AI model. An MCP server speaks JSON-RPC instead, a message format for calling remote functions. It uses tools/list to list its tools and tools/call to run one. A REST API speaks HTTP, with its own auth headers and status codes. The harness writes one adapter for each outside source. An MCP adapter maps tools/call replies into a ToolResult. It turns isError, JSON-RPC errors and timeouts into error results. An HTTP adapter maps 4xx and 5xx responses (the HTTP error codes) the same way. Each adapted tool carries its server name as a prefix, such as mcp__github__create_issue. So two servers cannot clash on the same name. The loop, the permission gate and the transcript never learn where a tool really lives.

class McpToolAdapter(Tool):
    def __init__(self, client, srv, spec):
        self.client, self.spec = client, spec
        self.name = f'mcp__{srv}__{spec.name}'
    def run(self, args):
        try:
            r = self.client.call_tool(
                self.spec.name, args)
        except McpError as e:
            return ToolResult.error(str(e))
        return ToolResult(
            content=r.content,
            is_error=r.isError)

Be carefulSome adapters turn remote failures into empty successes. This misleads the model. Pass errors and timeouts through as is_error results (results marked as failures).

Translate tool specs, calls and results between the harness's own types and each provider's wire format, in both directions.

Inside the harness, a tool is a ToolSpec, a call is a ToolCall and an output is a ToolResult. Each model provider wants something else on the wire, meaning in the exact format it sends and receives. One expects input_schema at the top level. Another puts parameters under a function key. Results go back as tool_result blocks with text or image content and an is_error flag. A provider converter owns all of these mappings in both directions. It turns a spec into the wire format. It turns a tool_use block into a ToolCall, and back again when the history is sent again. It turns a ToolResult into content blocks. Adding a provider then touches only one class. Round-trip tests (converting out and back again) can prove that nothing is lost.

class MessagesConverter:
    # spec_out(): ToolSpec -> tools[] entry
    def call_in(self, b):
        return ToolCall(b.id, b.name, b.input)
    def call_out(self, c):  # history replay
        return dict(type='tool_use', id=c.id,
            name=c.name, input=c.args)
    def result_out(self, r):
        return dict(type='tool_result',
            tool_use_id=r.call_id,
            content=r.blocks,
            is_error=r.is_error)

Be carefulConverters break quietly when a provider adds a new block type. Fail loudly on unknown types instead of dropping them.

Give the model one simple tool that hides a many-step system, so the model makes one call instead of five.

Running a project's tests in a fresh sandbox (a sealed space where code runs safely) takes many steps. The harness must start a container, install dependencies and start the test runner. Then it must read a JUnit XML report (a standard test results file) and trim megabytes of logs. If these were five tools, the model would spend turns and context on running them. The context is the text the model can see at one time. The model could also run them in the wrong order. A run_tests facade takes one optional path filter. It does every step inside and returns a short summary of failures. The parts are still there for harness code and other facades to use. But the model's tool list stays small.

def run_tests(path='.'):
    box = sandbox.acquire(repo)
    deps.install(box, lockfile='uv.lock')
    raw = runner.run(box, 'pytest', path)
    rep = junit.parse(raw.xml)
    return summarise(rep.failures,
                     max_tokens=800)

Be carefulA facade that hides too much leaves the model stuck when something unusual breaks. Include a log path so it can dig deeper.

Keep one map from names to tools, which the harness uses to list tools for the model and to route each tool_use call.

Three kinds of tools go into one ToolRegistry at startup. These are built-in tools, tools adapted from MCP servers and plugin tools. MCP is a standard way to connect tools to an AI model. Each tool is stored under a unique name such as read_file or mcp__github__create_issue. The registry produces the tools array sent with each request. A tool_use block is the model's request to run a tool. When one arrives, the dispatcher (the harness code that routes calls) looks up block.name and calls that handler. Duplicate names fail when they are registered, not later when they are called. An unknown name returns an is_error tool_result (a result marked as a failure), so the model can correct itself.

class ToolRegistry:
    def __init__(self):
        self._tools = {}
    def register(self, tool):
        if tool.name in self._tools:
            raise DuplicateTool(tool.name)
        self._tools[tool.name] = tool
    def get(self, name):
        return self._tools.get(name)
    def schemas(self):
        return [t.spec() for t in
                self._tools.values()]

Be carefulChanging the tools array during a session breaks the prompt cache (saved work the provider can reuse when a request starts the same way). This is because tools sit at the start of the cached prefix. Freeze the list for each session. Bring in late tools through deferred loading (adding them later without changing the start of the request).

Create tools from a name and a config entry, so agent definitions list tools as data instead of building them in code.

An agent definition file says tools: [bash, web_fetch, read_file]. It also has settings for each tool, such as a timeout or a list of allowed domains. A ToolFactory maps each name to a constructor (the code that builds that tool). It injects shared dependencies, meaning it hands each tool the shared parts it needs. These include the sandbox handle, the HTTP client and credentials. Then it returns ready Tool objects. Subagent definitions, eval configs and user settings all go through the same factory. So a tool is always built the same way, with the same safety defaults.

class ToolFactory:
    def __init__(self, deps):
        self.deps = deps
    def create(self, name, cfg):
        if name == 'bash':
            return BashTool(
                self.deps.box, cfg.timeout)
        if name == 'web_fetch':
            return WebFetch(
                self.deps.http, cfg.domains)
        raise UnknownTool(name)

Be carefulSometimes creating a tool is one constructor call with no shared dependencies. Then a factory only adds an extra step. Call the class directly.

Let a base runner own the agent workflow, while each subclass decides which executor or tool it creates.

A BaseRunner runs the turn loop, the approval flow and the writing of the transcript (the saved record of the conversation). Whenever it needs a command executor, it calls self.make_shell(). LocalRunner overrides make_shell, meaning it replaces it with its own version. Its version returns a shell on the user's machine. CloudRunner returns a shell inside a container with no network access. Tests make one more subclass. It returns a fake shell that replays recorded output. The loop is written and tested once, and only the creation step changes.

class BaseRunner:
    def run(self, task):
        shell = self.make_shell()
        return self.loop(task, shell)
    def make_shell(self):
        raise NotImplementedError

class LocalRunner(BaseRunner):
    def make_shell(self):
        return LocalShell(cwd=self.repo)

class CloudRunner(BaseRunner):
    def make_shell(self):
        return SandboxShell(net=False)

Be carefulOne subclass for each variation adds up fast. When the choice is plain config, pass in a factory function instead.

Create a whole matching family of tools, such as a shell and a file system, for one environment, so local and sandboxed tools never mix.

A coding agent needs a shell, a file system tool and often a browser. They must agree on where the workspace lives. A ToolkitFactory interface declares make_shell, make_fs and make_browser. LocalToolkit returns tools that act on the user's disk. SandboxToolkit returns tools that all target the same container and mounted volume. A sandbox is a sealed space where code runs safely. The session picks one factory from its environment setting. So a file the sandbox file system writes is always visible to the sandbox shell. And no code path can pair a local shell with a sandboxed file system.

class ToolkitFactory(Protocol):
    def make_shell(self): ...
    def make_fs(self): ...

class SandboxToolkit(ToolkitFactory):
    def __init__(self, box):
        self.box = box
    def make_shell(self):
        return ContainerShell(self.box)
    def make_fs(self):
        return VolumeFS(self.box.mount)

kit = KITS[session.env](env_handle)
tools = [kit.make_shell(), kit.make_fs()]

Be carefulAdding a new kind of tool means changing every factory in the family. So keep the family small and stable.

Build a frozen kit of named tool builders once at startup, then make exactly the tool set each agent profile asks for.

At startup, the harness fills a ToolKit with builder functions. bash builds a sandboxed shell. grep builds a search tool. read builds a file reader that can only see the workspace. The kit is then frozen. Agent profiles such as reviewer.yaml list only names. kit.create(['grep', 'read']) returns a fresh tool set for that subagent (a helper agent for one task). The kit is closed after startup. So no profile can reach a tool the operator did not install. Each subagent also gets new tool instances, not shared ones that others can change.

kit = ToolKit()
kit.add('bash', lambda: Bash(box=sandbox))
kit.add('grep', lambda: Grep(root=repo))
kit.add('read', lambda: ReadFile(root=repo))
kit.freeze()

profile = load('agents/reviewer.yaml')
tools = kit.create(profile.tools)
# -> [Grep(...), ReadFile(...)]
kit.create(['curl'])  # raises UnknownTool

Be carefulA builder might return one shared object instead of a new one. Then subagents overwrite each other's state.

Put together a large model request piece by piece, from system blocks, tools, history, cache breakpoints and limits, then check it once.

One model request can hold many parts. It can have a system prompt in layers, dozens of tool schemas and a long message history. It can have cache_control breakpoints (markers for where the prompt cache should save work). It can also have a thinking budget and max_tokens (a cap on reply length). A RequestBuilder lets the loop add these parts as they become known. System blocks come from settings and memory files. Tools come from the registry. History comes from the transcript after compaction (shrinking old conversation to save space). Then build() checks the whole request once. The total tokens must fit in the context window (the text the model can see at one time). The cache breakpoints must stay within the provider's limit. Every tool_result must be paired with its tool_use. Subagent and eval versions reuse the same builder with different parts.

req = (RequestBuilder(model=MODEL)
       .system(BASE_PROMPT)
       .system(memory.text(), cache=True)
       .tools(registry.schemas())
       .messages(transcript.compacted())
       .max_tokens(8000)
       .build())  # validates budget + pairs
resp = client.messages.create(**req)

Be carefulIf build() does no checking, the builder is only a long constructor. Put the checks across fields there, or skip the pattern.

Force a subagent or request spec through its required steps in a set order, so build() is out of reach until each is set.

Starting a subagent (a helper agent for one task) needs a model, a system prompt, a tool set and a token budget. A forgotten budget means a run that never stops. A SubagentSpec step builder starts with a ModelStep that has only .model(). That returns a PromptStep with only .system(). Next comes a ToolsStep with only .tools(), then a BudgetStep with only .budget(). Only the final BuildStep offers optional extras and .build(). A type checker (a tool that checks code types before it runs) flags a missing step before the harness runs. The order matches how people review specs: model, instructions, abilities, limits.

spec = (SubagentSpec.new()
        .model('fast-model')
        .system(REVIEW_PROMPT)
        .tools(['grep', 'read'])
        .budget(tokens=50_000)
        .build())

class ModelStep:
    def model(self, m) -> 'PromptStep':
        return PromptStep(m)
# ModelStep has no .build(): skipping fails

Be carefulA step builder adds one class for each step. So keep it for specs where a missing field is dangerous, not for every options object.

Copy a tuned agent or tool config template and change only a few fields, instead of building each version from scratch.

A well-tuned code reviewer subagent (a helper agent for one task) has a long system prompt and a chosen tool list. It also has a model choice, stop rules and hooks (code that runs at set moments). The orchestrator is the main agent that hands out work. To send reviewers over several folders, it deep-copies the reviewer template. A deep copy copies everything inside, including nested lists. Then it sets only the folder scope and output file on each copy. Put those changes in the first user message, not the system prompt. Then every clone keeps the template's tools and system prompt exactly the same, byte for byte. So all clones can reuse one prompt cache entry (saved work the provider can reuse when a request starts the same way). But that entry exists only once the first request starts to respond. So start one clone before the rest. The template itself is never changed. The next batch of copies starts from the same known state.

base = templates.get('reviewer')

def variant(scope):
    spec = copy.deepcopy(base)
    spec.scope = scope
    spec.output = f'reviews/{scope}.md'
    return spec

specs = [variant(d) for d in
         ['api/', 'web/', 'db/']]
for s in specs: harness.spawn(s)

Be carefulA shallow copy shares nested lists such as tools or hooks. So one copy's edit leaks into the template and the other copies. Always deep-copy.

Put a permission and sandbox layer with the same interface in front of every real tool, so nothing runs without passing its checks.

Every tool in the registry is wrapped in a GuardedTool proxy. It offers the same run(args) as the real tool. Before passing a call on, the proxy checks allow and deny rules against the tool name and arguments. Examples are writes outside the workspace, or network calls to hosts not on the list. When no rule decides, it shows an approval card to the user. Then it runs the real call inside the sandbox (a sealed space where code runs safely), with a timeout. Rules that match shell command text are easy to get around. So they only decide what needs approval, and the sandbox stays the hard boundary. On the way back, the proxy removes secrets from the result. It writes the decision to an append-only audit log (records can be added but never changed), and also to the transcript. The agent loop cannot tell the difference. And there is no path to the real tool that skips the proxy.

class GuardedTool(Tool):
    def __init__(self, real, policy, ui):
        self.real, self.policy = real, policy
        self.ui, self.name = ui, real.name
    def run(self, args):
        v = self.policy.check(self.name, args)
        if v == 'ask':
            v = self.ui.approve(self.name, args)
        if v != 'allow':
            return ToolResult.denied(v)
        out = sandbox.run(self.real, args)
        return redact(out)

Be carefulA proxy only guards what it wraps. A tool registered without it, or a raw subprocess call elsewhere in the harness, is an open side door.

Use a light stub in place of a costly tool, and load its full schema or start its server only when first needed.

A session may have hundreds of MCP tools available. MCP is a standard way to connect tools to an AI model. Putting every full JSON schema (the full description of a tool's inputs) in the system prompt would cost tens of thousands of tokens. It would also start servers that are never used. So the harness registers a stub for each tool. A stub is a light stand-in that holds only the tool's name and a one-line description. The model may pick a deferred tool (one not fully loaded yet) for the first time. It may do this through a tool search call, or directly. Then the stub starts the MCP server and fetches the full schema. It adds the schema to the context and swaps in the real adapter. Later calls go straight through.

class LazyTool(Tool):
    def __init__(self, name, desc, loader):
        self.name, self.desc = name, desc
        self._load, self._real = loader, None
    def real(self):
        if self._real is None:
            self._real = self._load()
            ctx.add_schema(self._real.spec)
        return self._real
    def run(self, args):
        real = self.real()
        if not real.spec.valid(args):
            return ToolResult.retry(real.spec)
        return real.run(args)

Be carefulLoading a schema during a session changes the prompt. So add new schemas after the cached prefix (the start of the request that the prompt cache saves). Do not reorder the tool list.

Build tool stand-ins while the program runs, from JSON schemas, and send every call through one shared handler.

MCP is a standard way to connect tools to an AI model. When an MCP server connects, the harness asks it for tools/list (its list of tools). The server answers with each tool's name, description and JSON input schema (a description of the allowed arguments). But the harness has no Python class written for any of these tools. So a proxy factory reads each schema and builds a stub on the spot. A stub is a small stand-in function with the right name, help text and argument checks. Every stub passes its call to one shared invoke handler. That handler packs the arguments into a tools/call request and sends it to the server over the connection. It then turns the reply into a tool_result block (the message that gives the model the tool's answer). So connecting a new server adds new tools without anyone writing or deploying code. The tool list can also change later. When the server sends notifications/tools/list_changed, the harness fetches tools/list again and rebuilds the stubs.

def invoke(client, schema, args):  # handler
    validate(args, schema['inputSchema'])
    return client.call(schema['name'], args)

def make_stub(client, s):
    stub = lambda **a: invoke(client, s, a)
    stub.__name__ = s['name']
    stub.__doc__ = s['description']
    return stub

for s in client.list_tools():
    tools[s['name']] = make_stub(client, s)

Be carefulGenerated stubs trust whatever the server sends. A hostile server can hide instructions to the model inside tool names or descriptions. So check names and descriptions against an allowlist (a list of approved items) before you show them to the model.

Wrap a tool in layers such as logging, retries, rate limits and telemetry. Each layer adds one job behind the same interface.

The harness keeps each tool as a plain object with a run(args) method. When it builds the tool set, it wraps each tool in layers. A Telemetry wrapper records a span (a timed trace record) and the call's cost. A Logged wrapper writes the call and its result to the transcript (the record of the session). A Retry wrapper tries again after transient errors (short-lived failures), waiting longer each time. This waiting is called backoff. It also follows any Retry-After hint from the server, which says how long to wait. A RateLimit wrapper waits for a free slot before calling a paid API. Every wrapper shows the same name and schema as the real tool. So neither the model nor the executor (the part that runs tools) can tell how many layers there are. Configuration chooses the layers for each tool. A local file read might get only logging. Web search gets all four. A tool with side effects, such as create_ticket, gets no Retry. The exception is when it sends an idempotency key. That is a unique ID that lets the server spot a repeat and do the work only once.

class Retry(Tool):
    def __init__(self, inner, n=3):
        self.inner, self.n = inner, n
        self.name = inner.name
    def run(self, args):
        for i in range(self.n):
            try: return self.inner.run(args)
            except Transient as e:
                sleep(e.retry_after or 2 ** i)
        raise ToolFailed(self.name)

t = Telemetry(Logged(Retry(RateLimit(web))))

Be carefulOrder matters. A retry outside the rate limiter waits for a new slot on each attempt. A retry inside it can burst past the limit. So set the order of layers in one place.

A tool hands the real work to a helper object or a subagent, then returns that answer as its own result.

A research tool that the model sees may contain no search code at all. It holds a link to a delegate (the helper that does the work) and passes the request on. The heavier version hands the work to a subagent (a second agent working for the first). The tool starts a fresh agent with its own context window (the text the model can see at one time). The subagent gets a smaller set of tools and a written brief. The tool waits for the subagent's final report. It then returns that report as the tool_result (the message that gives the model the tool's answer). The main agent's context sees one call and one answer. It does not see the dozens of tool calls the subagent made along the way. You can swap the delegate, for example for a cheaper model or a different search service. Nothing changes for the caller.

class ResearchTool(Tool):
    name = 'research'
    def __init__(self, spawn):
        self.spawn = spawn      # the delegate
    def run(self, args):
        sub = self.spawn('researcher',
                tools=['search', 'read'])
        report = sub.run(brief(args['q']))
        return report.summary

Be carefulDelegation hides cost. A one-line tool call can start a subagent that spends thousands of tokens (the units models read and write). So pass a token budget down and report usage back up.

Put shared chores, like checking paths or shortening results, in one helper that many tools call, not a copy per tool.

Many tools need the same chores done to what goes in and what comes out. A PathGuard servant takes a requested file path and works out the real full path. It follows symlinks (shortcuts that point to other files). It rejects anything outside the workspace root (the top folder the agent may use). read_file, write_file and grep all call it. A ResultTruncator servant cuts long tool output down to a token budget (a size limit in tokens, the units models read). It keeps the start and the end. It adds a note telling the model how to ask for the next part. The tools hand their data to the servant. They do not inherit from a shared base class. So the rule lives in one place, and a security fix reaches every tool at once.

class PathGuard:                 # servant
    def __init__(self, root):
        self.root = root.resolve()
    def safe(self, p):
        r = (self.root / p).resolve()
        if not r.is_relative_to(self.root):
            raise Denied(p)
        return r

def read_file(args, guard, trunc):
    p = guard.safe(args['path'])
    return trunc.fit(p.read_text(), 8000)

Be carefulSometimes a servant grows branches for single tools, such as if tool == 'grep'. Then it has become a god object (one object that does too much). Keep it to the shared rule and let each tool handle its own quirks.

When a feature is switched off, plug in a helper that does nothing, so the agent loop never checks for None.

The agent loop calls many optional helpers: a tracer (which records what happens), a memory store, a UI renderer (which draws the screen) and an approval prompter. Sometimes the agent runs headless, with no screen, in a scheduled task. There is no chat UI then. So the harness plugs in a NullRenderer whose on_token and on_tool_card methods do nothing. When long-term memory is off, a NullMemory returns an empty list on recall. It accepts writes but stores nothing. When there is no telemetry exporter (the part that sends trace data out), a NullTracer hands out empty spans (empty trace records). The loop calls the same methods every time. There are no 'if renderer is not None' checks spread through the code.

class NullMemory(Memory):
    def recall(self, query): return []
    def remember(self, fact): pass

class NullTracer(Tracer):
    def span(self, name): return nullcontext()

mem = VectorMem(url) if url else NullMemory()
tr = OTelTracer() if otel else NullTracer()

Be carefulNever give the model a tool that silently does nothing. A write_file that quietly does nothing lets the agent believe it saved its work. If a tool is missing, leave it out of the tool list instead.

For expected odd outcomes, like no matches or file not found, return a special result object that knows how to behave.

Search and lookup tools hit dead ends you can predict. grep finds nothing, a path does not exist, or a ticket ID is unknown. The tool does not return None or raise an error (throw an exception). Instead it returns a special case object such as NoMatches or MissingFile. That object has the same ToolResult interface as any other result. Its to_block() method makes a useful tool_result (the message that gives the model the tool's answer). For example: 'No matches for retry_cfg in 412 files. Try a broader pattern.' It keeps is_error false, because an empty search is a valid answer. It costs only a few tokens. The executor, the transcript and the renderer treat it like any other result.

class NoMatches(SearchResult):
    def __init__(self, pat, n):
        self.pat, self.n = pat, n
    def to_block(self):
        return text(
            f'No matches for {self.pat} in '
            f'{self.n} files. Try a broader '
            'pattern or check the spelling.')

hits = index.search(q)
return Hits(hits) if hits else NoMatches(q, n)

Be carefulUse special cases only for expected situations. A disk error or a permission failure is a real error. Show it as one. Do not dress it up as 'no results'.

Let harness parts ask a tool, while running, for optional extras such as streaming, undo or preview, without enlarging every tool's interface.

Every tool has a small core interface: name, schema and run. But some tools can do more. A shell tool can stream its output line by line. An edit tool can show a diff preview (the changes, before it applies them). It can also keep an undo record afterwards. A browser tool can return a screenshot. The harness does not add stream(), preview() and undo() to every tool. Instead it calls tool.get_extension('preview'). It gets back either an object that offers Previewable or None. The approval card shows a diff when a preview exists. If none exists, it shows the raw arguments. The streaming renderer listens only to tools that return a Streaming extension.

class EditTool(Tool):
    def __init__(self):
        self.ext = {'preview': DiffPreview(),
                    'undo': EditUndo()}
    def get_extension(self, name):
        return self.ext.get(name)

pv = tool.get_extension('preview')
if pv: ui.show_diff(pv.render(args))
else:  ui.show_args(args)

Be carefulIf every tool ends up with the same extension, move it into the core interface. Extension lookups are for features that are truly optional.

Tag tools as read-only, destructive or approval-required with empty markers. The permission gate checks them before running anything.

Tool classes inherit from empty marker types such as ReadOnly, Destructive or NeedsApproval. Or they carry matching flags, like readOnlyHint and destructiveHint, in an MCP tool's annotations. (MCP is a standard way to connect tools to an AI model.) The markers add no methods. The permission gate checks them before each call. Read-only tools run at once, even in plan mode. Destructive ones pause and wait for the user's approval. Network tools are blocked in an offline sandbox. The scheduler reads the same tags. It runs read-only calls in parallel and runs anything that writes one at a time.

class ReadOnly: pass       # markers carry
class Destructive: pass    # no methods

class ListDir(Tool, ReadOnly): ...
class DeleteFile(Tool, Destructive): ...

def gate(tool, mode):
    if isinstance(tool, Destructive):
        return 'ask'
    if isinstance(tool, ReadOnly):
        return 'run'
    return 'run' if mode == 'auto' else 'ask'

Be carefulA marker is only a claim made by the tool's author. Do not trust the hints on third-party MCP tools. Keep your own policy list for them.

Define each tool as a data record (name, schema, handler) instead of a new class, so adding a tool adds no code.

An MCP tool definition is a type object. (MCP is a standard way to connect tools to an AI model.) It is a JSON record with name, description, inputSchema (the allowed arguments) and annotations. Every server's tools are instances of one ToolDef type, not separate classes. The harness has one general ToolCall class. Each ToolCall points at its ToolDef for checking and display, the way a game figure points at its card. Teams add tools by editing a JSON or YAML file (simple text formats for data). The file links a definition to an existing handler, such as a web endpoint, a shell command template or an MCP call. The harness loads these files at startup. Settings shared by every call of a type, such as the default timeout and cost class, live on the definition.

class ToolDef:             # the type object
    def __init__(self, d):
        self.name = d['name']
        self.schema = d['inputSchema']
        self.timeout = d.get('timeout', 30)
        self.handler = HANDLERS[d['handler']]

class ToolCall:            # an instance
    def __init__(self, tdef, args):
        validate(args, tdef.schema)
        self.tdef, self.args = tdef, args

Be carefulTools defined as data are only as safe as the handlers they can point at. Keep the handler table small and reviewed. Do not let settings files reach any code they like.

The model writes a short program in a limited action language. A small harness VM checks it, then runs it step by step.

The harness does not give the model a general shell. Instead it defines a tiny action language for browsing: OPEN url, FIND selector, CLICK ref, TYPE ref text, EXTRACT field. Each of these commands is an opcode. The model writes a short program as a JSON list of instructions inside one tool_use (its request to run a tool). A harness VM (a small virtual machine) checks the program before it runs. It allows only known opcodes and correct argument types. It allows at most 50 steps and only approved domains (websites). Then it runs the program with a program counter (which tracks the current step) and a step budget. The instruction set has no opcode for running any code or touching files. The VM also checks the website again at every page change. So a program can do only what those opcodes allow, and only on approved sites. That still includes submitting forms. So CLICK and TYPE on pages with side effects need an approval step or a read-only session. The program is fixed before any page is read. So text hidden in a page (a prompt injection) cannot add steps. And putting ten steps into one program saves nine round trips to the model.

OPS = {'OPEN': op_open, 'CLICK': op_click,
       'TYPE': op_type, 'EXTRACT': op_extract}

def run(program, limit=50):
    verify(program, OPS, limit)   # first
    trace = []
    for pc, ins in enumerate(program):
        out = OPS[ins['op']](*ins['args'])
        trace.append((pc, ins['op'], out))
        if out.failed: break
    return trace

Be carefulEvery opcode you add widens what a model, or a prompt injection, can make the VM do. An EVAL or SHELL opcode turns the design back into running any code at all.

Keep a tool call's extra details in one flexible key-value document. Small typed views read only what each part needs.

A tool call collects metadata (details about the call) from many places. Some comes from the model's input, and some from MCP annotations. Some comes from harness settings, such as timeout and cost class. Some comes from UI hints, such as an icon and display title. Audit fields such as approved_by and duration_ms are added along the way. The harness does not use one class with dozens of optional attributes. Instead it keeps them all in a single dict-backed document (built on a key-value dictionary) with nested child documents. The model's arguments always sit in their own input child. So they cannot shadow (hide or replace) a harness key. Small typed views, such as HasTimeout, HasApproval and HasDisplay, give each part simple ways to read the same dict. So the executor, the permission gate and the tool card renderer each read only what they understand. Only the part that owns a trust-bearing key may write it. approved_by is one such key. Unknown keys from a newer MCP server pass through to the transcript untouched.

class Document(dict):
    def child(self, k):
        return Document(self.get(k, {}))

class HasTimeout(Document):
    def timeout(self):
        return self.get('timeout_s', 30)

class HasApproval(Document):
    def approver(self):
        return self.get('approved_by')

class CallDoc(HasTimeout, HasApproval): pass

Be carefulA bag with no schema lets typos such as 'timout_s' fail without any error. Give each view a table of defaults. Log any key that no view ever reads.

Give each tool a settings map that falls back to a parent map. Tools inherit defaults and change only what differs.

Tools share many settings: timeout, retry count, output token limit, sandbox profile and approval mode. The harness keeps a root map of defaults. Each MCP server gets a child map. For example, a slow remote server sets timeout_s to 120. Each tool gets a map that holds only its own differences. A lookup such as tool.get('timeout_s') checks the tool, then its server, then the root. It returns the first value it finds. Changing a root default updates every tool that has not set its own value. A subagent's tool set can be a child map that overrides approval_mode. But it may only make it stricter. So a child never loosens what its parent requires.

class Props:
    def __init__(self, parent=None, **own):
        self.parent, self.own = parent, own
    def get(self, k):
        if k in self.own: return self.own[k]
        if not self.parent: raise KeyError(k)
        return self.parent.get(k)

root = Props(timeout_s=30, approval='ask')
remote = Props(root, timeout_s=120)
fetch = Props(remote, approval='auto')
fetch.get('timeout_s')    # 120, from remote

Be carefulLong chains make 'why is this timeout 120?' hard to answer. Keep the chain short. Have the tool inspector show which level supplied each value.

Bundle a tool's many arguments into one typed request object. Check it once and pass it along whole.

A run_command tool can take a command, a working folder, environment variables, a timeout, stdin (input fed to the command), a sandbox profile and an output limit. Passing seven loose arguments through the permission gate, the sandbox and the executor invites mix-ups. So the harness parses the tool_use input once into a typed ShellRequest. This is a dataclass or Pydantic model (a Python class that checks data types), built from the same JSON schema the model sees. Checks and defaults happen in one place. The gate checks req.command and req.cwd. The transcript stores the request whole. Adding a field later changes the request class, not every function along the way.

class ShellRequest(BaseModel):
    model_config = ConfigDict(
        frozen=True, extra='forbid')
    command: str
    cwd: Path = Path('.')
    timeout_s: int = Field(60, le=600)
    env: dict[str, str] = {}
    max_output: int = 30_000

req = ShellRequest.model_validate(block.input)
gate.check(req)    # command, cwd, env keys
result = sandbox.run(req)

Be carefulDo not let one request class turn into a grab bag shared by unrelated tools. One request type per tool keeps the schema the model sees honest.

Run every tool inside a function that sets up the sandbox first and always cleans it up afterwards, even on failure.

Running a code tool needs setup. It needs a scratch folder and a container or a slice of a VM (an isolated mini computer). It needs the workspace files mounted (made available inside), a network policy, and limits on CPU and clock time. Afterwards, the container must be stopped, temporary files deleted and logs collected. The harness offers one context manager, sandbox(profile), that does all of this. (A context manager is a Python block that runs setup at the start and cleanup at the end.) It runs the tool's body in the middle. Tool authors write only the body. A crash, a timeout or the user pressing stop still reaches the finally block (the cleanup code that always runs). So leftover containers do not pile up between turns.

@contextmanager
def sandbox(profile):
    box = containers.start(profile)
    try:
        box.mount(workspace, ro=profile.ro)
        yield box
    finally:
        box.collect_logs(transcript)
        box.kill(); box.cleanup()

with sandbox(DEFAULT) as box:
    out = box.exec(req.command, timeout=60)

Be carefulCleanup cannot depend on the body behaving well. Enforce time limits from a watchdog (a separate timer process) outside the sandbox. A body that hangs never hands control back to the finally block.

Set up an agent with chained method calls that read like a sentence. Each call returns the builder for the next.

Agent SDKs (software kits for building agents) often offer a fluent setup API: Agent('reviewer').model(...).system(...).tools(read, grep).max_turns(20).on_tool_use(audit).build(). Each method records one setting and returns the builder. So a whole agent definition fits in one readable expression. It is easy to review in a pull request (a proposed code change). The final build() checks that the settings fit together. For example, a shell tool must have a sandbox, and the token budget must fit the model's context window (the text the model can see at one time). Then build() returns an AgentConfig that cannot be changed. Subagent definitions start from a shared base chain and add a few calls.

class Agent:
    def model(self, m):
        self.cfg['model'] = m
        return self          # enables chaining

cfg = (Agent('reviewer')
       .model('large')
       .system(load('review.md'))
       .tools(read_file, grep, git_diff)
       .max_turns(20)
       .build())           # validates once

Be carefulChains hide the in-between state. A setting given twice, or in a conflicting order, can silently win. Make build() reject contradictions instead of keeping the last call.

Send every model call through one client that holds the keys, picks a provider and switches over when one is down.

The agent loop (the code that runs the model turn after turn) never imports a provider SDK (the provider's own code library). It calls gateway.complete(request). The gateway adds API keys (secret passwords for the provider) from the secret store. It turns the request into the provider's wire format (the exact shape that provider expects). It then routes the request by model alias, region or tenant policy. A model alias is a short name, like 'large', that points to real models. A tenant is one customer company sharing the platform. The gateway also records usage and latency (how long each call takes) for every call in one place. Sometimes a provider returns errors or is marked unhealthy. Then the gateway sends the same request to the next configured route. It returns a response in one standard shape. So subagents (helper agents started by the main agent) and scheduled tasks all get the same behaviour.

class ModelGateway:
    def complete(self, req, tenant):
        routes = self.policy.allowed(
            tenant, self.routes[req.model])
        for p in routes:  # residency, vendor
            if not p.healthy():
                continue
            try:
                r = p.send(req, self.keys)
            except ProviderDown:
                p.mark_down(); continue
            self.meter.record(tenant, r.usage)
            return p.decode(r)
        raise NoProviderAvailable(req.model)

Be carefulA gateway might quietly switch to a weaker model. That can change how the agent behaves in the middle of a task. So log every reroute and record which model actually answered.

Run a sidecar, a helper beside the harness, that owns the model connection: keys, TLS encryption, retries, timeouts and telemetry.

The harness does not call the provider directly. It sends plain HTTP requests to a sidecar on localhost (the same machine). The sidecar runs in the same pod or container group (a set of containers deployed together). It holds the API key and handles TLS (the encryption on the connection to the provider). It adds retries with backoff (waiting longer before each new try) and enforces timeouts. It also sends out metrics and traces (records of each call's path and timing) for every model call. The sidecar does not care what language the caller uses. So a Python agent worker and a TypeScript UI server share one connection policy. To rotate keys or change egress rules (rules about which outside addresses traffic may reach), you redeploy the sidecar, not the agent code.

# harness: no keys, no retry logic
client = ModelClient(
    base_url='http://localhost:8787')
resp = client.messages.create(**req)

# ambassador sidecar, same pod
@app.post('/v1/messages')
async def relay(req):
    req.headers['x-api-key'] = vault.key()
    return await retry(upstream.post, req,
                       timeout=60)

Be carefulIf both the harness and the ambassador retry, one failure turns into many calls. Decide which side owns retries, and turn them off on the other side.

Keep the family of agent types apart from the family of model backends, so each side can grow without a class for every pair.

The agent side has its own family of classes. ChatAgent, CodingAgent and ResearchAgent differ in system prompt, tools and loop policy. The backend side has another family. A backend is the code that actually talks to a model. There is a hosted API backend, a cloud-platform backend and a local inference backend (running the model on your own machines). Each backend knows its own request format, streaming quirks and login method. Each agent holds a reference to a ModelBackend. It calls only that backend's narrow interface: generate, stream and count_tokens. Three agents and three backends need six classes, not nine combinations. A new backend works with every agent from day one.

class Agent:                 # abstraction
    def __init__(self, backend):
        self.backend = backend

class CodingAgent(Agent):
    def step(self, msgs):
        return self.backend.generate(
            CODE_PROMPT, msgs, TOOLS)

class LocalBackend(ModelBackend): ...
class HostedBackend(ModelBackend): ...

agent = CodingAgent(LocalBackend())

Be carefulKeep the backend interface to what every provider can support. If one provider's special feature leaks into it, the bridge quietly becomes an adapter for just one provider.

Re-send a model call that failed for a short-lived reason, waiting longer each time, and give up after a fixed limit.

Model APIs return transient errors (short-lived errors that may pass if you try again). A 429 rate_limit_error means you hit a usage limit. A 529 overloaded_error means the service is busy. There are also occasional 500 or 504 errors, and connections that drop mid-stream. A streaming call can also fail after a 200 (success) status, with an error event inside the stream. The client treats that error the same way as the matching HTTP error. If the response has a retry-after header (the provider's advice on how long to wait), the client follows it. Otherwise it waits using capped exponential backoff plus jitter. That means the wait doubles each time up to a maximum, plus a small random extra. Then it sends the identical request again. This is safe because a model call has no side effects (it changes nothing outside itself). Some errors will never succeed on a resend. Examples are 400 invalid_request_error, 401, 403, or a 429 caused by a used-up spending cap. These fail at once. The retry budget is limited by the number of attempts and by the turn's deadline. So a user in the chat never waits forever.

RETRYABLE = {408, 429, 500, 503, 504, 529}
for attempt in range(MAX_TRIES):
    try:
        return api.create(**req)
    except APIError as e:  # incl. SSE error
        if e.status not in RETRYABLE:
            raise
        if attempt == MAX_TRIES - 1:
            raise GaveUp(req) from e
        wait = min(CAP, BASE * 2 ** attempt)
        wait = random.uniform(wait / 2, wait)
        sleep(max(wait, e.retry_after or 0))

Be carefulRetries can be stacked at several layers, such as the SDK, the gateway and the agent loop. During an outage they multiply into dozens of calls. Retry in exactly one place.

When the preferred model cannot answer, give the same request to a substitute, so the turn still finishes, maybe at lower quality.

Sometimes retries run out, or a circuit breaker is open (a switch that blocks calls to a failing provider). Then the model client walks a fallback chain, which is an ordered list of substitutes. The chain is first filtered by the tenant's policy (the rules for that customer company). It might try the same model in another allowed region, then a smaller model. Last, it returns a typed 'busy, try again' status. Each step gets the same messages and tool definitions. They change only where the substitute needs it. Examples are a smaller max_tokens (output limit), no extended thinking, or dropping earlier thinking blocks. The substitute starts with a cold prompt cache. The prompt cache is saved work the provider can reuse when a request starts the same way, and cold means it is empty. So the substitute's first call pays the full input price. The harness tags the response with the model that actually served it. This keeps the transcript, cost records and evals (quality tests) honest. Some callers must use the strongest model, such as a review subagent. They opt out and fail loudly instead.

CHAIN = ['large', 'large-eu', 'small']

def complete(req, tenant):
    for model in tenant.allowed(CHAIN):
        try:
            r = call(model, fit(req, model))
            r.served_by = model
            return r
        except (Overloaded, BreakerOpen):
            continue
    return Degraded('busy, try later')

Be carefulA fallback model may not support the same tools, context length or output format. So run the agent's evals against every model in the chain.

Stop calling a provider that keeps failing, refuse calls quickly for a cool-down period, then let one test call decide whether to resume.

The model client keeps one breaker per provider, or per model and region. While the breaker is closed, calls flow normally. It counts 5xx (server error), 529 and timeout failures over a rolling window of recent time. It leaves 429s to rate limiting and retry-after. When failures cross a threshold, the breaker opens. For the next 30 seconds, every call fails at once with BreakerOpen. The fallback layer turns that into a switch to another model. So no agent wastes its retry budget on a dead endpoint. After the cool-down, the breaker goes half-open. It lets exactly one probe (test request) through. Success closes the breaker. Failure opens it again with a longer timer.

def call(self, req):
    with self.lock:          # one probe only
        if self.state == 'open' and \
           now() >= self.retry_at:
            self.state = 'half_open'
        elif self.state != 'closed':
            raise BreakerOpen(self.name)
    try:
        r = self.provider.send(req)  # 4xx ok
    except (ServerError, Timeout):
        self.fail()   # half_open -> open
        raise
    self.reset()      # closed, fails = 0
    return r

Be carefulCount only failures on the provider's side. A 400 error caused by a badly formed prompt is not the provider's fault. If those errors trip the breaker, a healthy provider goes offline for every agent.

Cap how many requests and tokens each caller may use per time window, and refuse the extra with a clear time to retry.

Providers already limit requests per minute, and input and output tokens per minute. Tokens are the small chunks of text that models read and write. When you go over, the provider answers 429 with a retry-after header (how long to wait). A multi-tenant harness serves many customer companies, called tenants. It adds its own limiter in front. This is usually a token bucket per organisation or per user. A token bucket is a counter that refills at a steady rate, and each call spends from it. The buckets live in a shared store. So one user's runaway agent loop cannot use up the shared provider quota. Before each call, the limiter charges the estimated input tokens plus the full max_tokens reservation (the most output the call may produce). This is on purpose, and it is stricter than the provider. The provider counts output as it is generated, and for most models it ignores cache reads. After the call, the limiter settles the real usage. A request that would overdraw the bucket is refused at once, with the time until enough tokens refill. A request that can never fit is refused outright.

class TokenBucket:
    def take(self, n):
        if n > self.capacity:
            raise TooLarge(n)   # never fits
        self.refill(now())
        if self.tokens < n:
            need = n - self.tokens
            raise RateLimited(need/self.rate)
        self.tokens -= n

# shared store, so all replicas agree
bucket = buckets.get(req.org_id)
bucket.take(estimate(req) + req.max_tokens)

Be carefulBad token estimates make the limiter useless or unfair. So compare against the usage the provider reports after every call.

Keep the harness's own agents within a safe share of the provider quota by pacing, capping, degrading or delaying their calls.

Every agent, subagent and scheduled job in the harness shares one provider quota. So the harness slows its own users before the provider starts returning 429 errors. A controller reads the rate-limit headers after each call. These headers say how many tokens remain and when the limit resets. As the remaining tokens shrink, it takes graded steps, one after another. First, a shared pacer spaces out calls. Next, fewer subagents may run at once. Then background work is degraded: shorter max_tokens, a smaller model, or no extended thinking. Last, the lowest-priority jobs are deferred (put off until later). Interactive turns, where a person is waiting, still go through, a little later. As headroom (spare quota) comes back, the harness lifts the steps in reverse order.

class Pacer:
    def adjust(self, h):
        left = int(h['tokens-remaining'])
        # reset is an RFC 3339 timestamp
        ts = parse_rfc3339(h['tokens-reset'])
        until = max(ts - now(), 0)
        self.gap = until / max(left / AVG, 1)

    async def wait_turn(self):
        slot = max(now(), self.next)
        self.next = slot + self.gap  # reserve
        await sleep(slot - now())

Be carefulThrottling only hides overload up to a point. If the ramp queue grows without end, you also need load shedding (dropping low-priority work) or backpressure (making the sender wait).

When demand is more than capacity, drop or delay the least important model work, so interactive turns keep getting answers.

Each model request carries a priority. It inherits that priority from the root of its run (the request that started the work). A user waiting in the chat UI is high priority, and so is any subagent working for that turn. A subagent on a background task is medium. Scheduled digests, eval runs and memory consolidation (tidying stored memories) are low. Sometimes provider errors rise, or the queue gets long. Or p95 latency (the time within which 95% of calls finish) crosses a threshold. Or the organisation's token budget is nearly gone. Then the admission check (the gate before each call) refuses calls below the current cut-off. It gives them a 'shed' status and moves them to a later scheduled task. Interactive turns stay fast. Shed work is recorded so it can run when load drops.

PRIORITY = {'chat': 3, 'subagent': 2,
            'eval': 1, 'digest': 0}

def admit(req):
    cutoff = 0
    if load.p95_ms > 8000: cutoff = 2
    elif load.queue > 200: cutoff = 1
    if PRIORITY[req.kind] < cutoff:
        deferred.put(req)   # run later
        raise Shed(req.kind)
    return dispatch(req)

Be carefulShedding must be visible. A background job that quietly gets dropped every afternoon looks like a bug. So show deferrals in logs and in job status.

Let a slow consumer push back on its producer, so work waits at the source instead of piling up in memory.

An orchestrator (the agent that hands out work) may fan out many subagents. It puts each task into a bounded queue, which has a fixed maximum size. Suppose every model-call slot in the dispatcher is busy. Then queue.put waits, and the orchestrator stops starting new subagents. So memory stays flat, however many tasks the model plans. Streaming to the chat UI works differently. If the harness pauses reading from the provider, the model does not slow down. A long pause may also make the provider close the stream. So the harness reads the provider stream at full speed. It writes into the transcript and into a bounded buffer for each client. When a slow mobile client falls behind, the harness coalesces (merges) the queued text pieces instead of holding up the provider. Wherever the producer can truly wait, the slowest stage sets the pace.

q = asyncio.Queue(maxsize=8)

async def orchestrator(plan):
    for task in plan.subtasks:
        await q.put(task)   # waits when full

async def dispatcher():   # start 4 of these
    while True:
        task = await q.get()
        try:
            await run_subagent(task)
        finally:
            q.task_done()

Be carefulAn unbounded queue anywhere in the chain breaks backpressure. The producer never feels the push, and memory grows until the process dies.

Put model jobs in a durable queue that a fixed pool of workers empties, so bursts become a steady, affordable call rate.

At 9am, hundreds of scheduled tasks start at once. A CI webhook (an automatic message from the build system) can launch fifty review agents in a minute. These do not each call the model right away. Instead, the harness writes a job record to a durable queue, one that keeps jobs safe if machines restart. It returns a run id. A pool of workers pulls jobs and runs the agent loop. The pool is sized to the provider's sustained rate limit (the steady rate it allows). So the provider sees a flat line of traffic, and the 429 errors stop. Callers watch the run status or get a notice when it is done. Nothing is lost if a worker restarts.

def on_webhook(event):
    run_id = f'pr-{event.pr}-{event.sha}'
    jobs.enqueue({'run': run_id,  # dedupe
                  'agent': 'pr_review',
                  'pr': event.pr})
    return {'run_id': run_id}   # at once

def worker():                    # N of these
    while job := jobs.receive(lease=900):
        if not runs.done(job['run']):
            result = run_agent(job)
            notify(job['run'], result)
        jobs.ack(job)  # redelivered if we die

Be carefulLoad leveling trades speed for stability. A user waiting in the chat should not sit behind a thousand batch jobs. So give interactive work its own lane.

Reuse work already paid for: cache the stable start of the prompt with the provider, and save repeatable tool results locally.

Every turn sends the tool definitions, system prompt and earlier conversation again. This stable part at the start is the prefix. The harness marks the end of the prefix with a cache_control breakpoint. The provider can then reuse its processed work when a later request starts with exactly the same prefix. The first call pays a cache-write premium, an extra charge above the normal input price. Later calls report cache_read_input_tokens. These cost a fraction of the normal input price, and the reply comes faster. The cache entry expires after about five minutes without a hit. A longer TTL (time to live) costs more to write. Prefixes shorter than a minimum length, which depends on the model, are not cached at all. The harness also has its own local cache. It memoises (saves the result of) read-only tool calls, such as fetching the same URL or reading an unchanged file. The cache key includes the tenant (customer company), the user, the tool name and the arguments. It also includes a content hash (a fingerprint of the content) or a short TTL. Anything with side effects or time-sensitive data skips the cache.

system = [{'type': 'text', 'text': RULES,
           'cache_control':
               {'type': 'ephemeral'}}]

@memo(key=lambda ctx, a: (ctx.tenant,
      ctx.user, a['url'], day()))
def fetch_page(ctx, a):
    return http.get(a['url']).text

r = client.messages.create(model=M,
    system=system, tools=TOOLS, messages=h)
log(r.usage.cache_read_input_tokens,
    r.usage.cache_creation_input_tokens)

Be carefulA timestamp or random id near the top of the system prompt changes the prefix on every call. Then every call pays the cache-write premium, and none ever gets a cheap read.

Keep costly resources such as sandboxes, browsers and connections ready, lend them to agents, and reset them when returned.

Starting a code sandbox container (a sealed space for running code) or a headless browser (a browser with no screen) takes seconds. Opening a fresh TLS connection (an encrypted network link) to the model API on every call also adds delay. So the harness keeps pools. There are a few warm (already started) sandboxes, a browser pool for web tools, and the HTTP client's keep-alive connection pool. An agent checks out a sandbox for a bash or code-execution tool call. When the sandbox comes back, the pool wipes the working folder, kills leftover processes and resets environment variables. Only then does it lend it again. When every instance is busy, callers wait with a timeout, or the pool grows up to a hard limit.

class SandboxPool:
    def acquire(self, timeout=30):
        sb = self.idle.get(timeout=timeout)
        return Lease(sb, self.release)
    def release(self, sb):
        try:
            sb.reset()  # wipe fs, procs, env
            self.idle.put(sb)
        except ResetFailed:
            sb.destroy(); self.spawn_one()

with pool.acquire() as sb:
    out = sb.run(tool_input['command'])

Be carefulA resource returned dirty can leak one session's files or cookies into another user's agent. So treat the reset as critical for security, and destroy anything that fails it.

Keep exactly one client per key, such as provider and model or tenant, created on first use and shared after that.

Each provider and model pair needs its own client. Each client has its own base URL, credentials, connection pool and rate-limit tracker. ModelClient.get(provider, model) returns the existing client for that key. On first use, it builds one. It does this under a lock (a guard that lets only one task in at a time). So two subagents running at the same time do not both build it. Every agent that asks for the same key shares one connection pool and one view of remaining quota. Another tenant's key (another customer company) gets a separate client with its own API key.

class ModelClient:
    _all = {}
    _lock = threading.Lock()

    @classmethod
    def get(cls, provider, model):
        key = (provider, model)
        with cls._lock:
            if key not in cls._all:
                cls._all[key] = cls(*key)
            return cls._all[key]

Be carefulA key might include a value that changes often, such as a per-request timeout. Then each call creates a new instance, and the multiton quietly becomes a memory leak.

Share one model client and settings object across the whole process, so every agent uses the same connection pool, credentials and limits.

A harness process usually creates one SDK client (the provider's code library) at startup. It shares that client across the agent loop, subagents and background workers. That one instance owns the HTTP connection pool (open network links kept for reuse). It also owns the API key, default timeouts and the retry policy. If each subagent built its own client, the process would open extra connections. Each copy would also track rate limits on its own. Settings loaded from environment variables and config files work the same way. get_settings() returns one cached object.

from functools import cache

@cache
def get_settings():
    return Settings.from_env()

@cache
def get_client():
    s = get_settings()
    return ModelClient(api_key=s.api_key,
                       max_retries=s.retries,
                       timeout=s.timeout)

Be carefulA singleton is global mutable state (shared data anyone can change) in disguise. Pass it into agents instead of calling get_client() deep inside tools. Otherwise tests cannot swap in a fake.

Read model output by taking what you need and ignoring the rest, fixing small mistakes instead of crashing on them.

A model asked for JSON (a common text format for structured data) does not always return clean JSON. It may wrap it in a markdown fence ( marks) or add a sentence before it. It may leave a trailing comma, return numbers as strings or invent an extra field. A tolerant reader in the harness removes fences and pulls out the first balanced object (from the first { to its matching }). It applies a gentle repair. Then it reads only the required fields, converts simple types and ignores unknown keys. The harness reads verdicts from grader models (models that score answers) and plans from planner subagents this way. Only when a required field is truly missing does the harness send a short correction request back to the model.

def read_verdict(text):
    raw = strip_fences(text)
    obj = repair_json(first_obj(raw))
    # take what we need, ignore the rest
    return Verdict(
        score=int(obj['score']),   # '7' ok
        reason=obj.get('reason', ''))

try:
    v = read_verdict(reply.text)
except (KeyError, ValueError):
    ask_model_to_fix(reply)

Be carefulTolerance belongs only at the edge where text is parsed. Pass strict, typed objects inward. Never 'repair' input that authorises a destructive tool action (one that deletes or changes things).

Deliver a model's answer piece by piece as it is written, or send only the fields a caller asked for.

With streaming on, the model API sends server-sent events (SSE: a stream of small messages over one connection). First comes message_start. Then each content block gets a content_block_start event. Next comes a run of content_block_delta events. These carry text_delta or input_json_delta pieces. Each block ends with content_block_stop. Then a message_delta event carries the stop_reason (why the model stopped) and usage. Last comes message_stop. The harness sends text pieces to the chat UI at once. It holds input_json_delta pieces until a tool_use block (the model's request to run a tool) is complete. It runs tools only after the stream ends with stop_reason tool_use. It can cancel the stream when the user presses stop. Field selection is the same idea for reads. The UI polls a status endpoint. That endpoint returns only the run's state and token count, never the whole transcript.

with client.messages.stream(**req) as s:
    for ev in s:
        if ev.type == 'content_block_delta' \
           and ev.delta.type == 'text_delta':
            ui.send(ev.delta.text)
        # input_json_delta: SDK buffers it
        # ping and unknown events: ignore
    msg = s.get_final_message()
if msg.stop_reason == 'tool_use':
    run_tools(msg.content)  # whole inputs
elif msg.stop_reason == 'max_tokens':
    continue_or_fail(msg)   # may be cut off

Be carefulNever act on a partial tool input. A half-streamed path such as '/home/user/pro' is a valid start of the real path, and it points to the wrong file.

Store tokens and cost as typed amounts with a unit or currency, so sums, budgets and reports cannot mix numbers that do not match.

Each response has a usage block with four counts. input_tokens is only the uncached part of the prompt. cache_creation_input_tokens counts tokens written to the prompt cache. cache_read_input_tokens counts tokens read from it. output_tokens counts what the model wrote. Each kind has its own price for each model, and cache writes are priced by TTL (how long the cache entry lives). Cached tokens are billed separately, so the cost ledger must include all four counts. As bare integers, these counts invite bugs. Code might treat input_tokens as the whole prompt, or add output tokens to input tokens. It might add a large model's tokens to a small model's as if they cost the same. So the harness uses value objects: Tokens(n, kind, model) and Money(pico, 'USD'). It converts between them through an explicit, versioned price table. It does arithmetic in whole-number units small enough to price one cached token exactly. These are pico-dollars (millionths of a millionth of a dollar). With streaming, it takes usage from the final message_delta. Those counts are cumulative (running totals), so it does not add up every event. Budgets for a session, a subagent tree or a tenant (customer company) are Money too. So a check like spent + estimate > budget is type-safe (the types stop mistakes).

@dataclass(frozen=True)
class Money:
    pico: int; cur: str = 'USD'  # 1e-12
    def __add__(s, o):
        assert o.cur == s.cur
        return Money(s.pico + o.pico, s.cur)
def cost(u, model):
    p = PRICES[model]   # pico-USD per token
    n = (u.input_tokens * p.inp
       + u.cache_creation_input_tokens * p.wr
       + u.cache_read_input_tokens * p.rd
       + u.output_tokens * p.out)
    return Money(n)     # exact, round later

Be carefulFloating-point dollars (decimals that computers store inexactly) drift after millions of tiny per-call charges. Store whole-number units smaller than a millionth of a dollar, and round only when displaying or invoicing.

Give the agent's memory a simple list-like interface (add, search, remove), so loop code never talks to the database directly.

The agent loop (the code that runs the model turn after turn) asks a MemoryRepository for what matters to this turn: repo.search(query, k=5). Here k is the number of results. Behind the repository there might be a vector store (a database that finds text by meaning). It uses embeddings, which are lists of numbers that stand for meaning. It could also be a SQL table of past chats or a folder of markdown memory files. The repository speaks in the harness's own objects, such as Memory and Turn. It applies scope, such as the user id and how recent a memory is. It also hides the embedding call. You can swap the vector store, or use a simple in-memory repository in evals (automated test runs), without changing any loop code.

class MemoryRepo(Protocol):
    def add(self, m: Memory): ...
    def search(self, q, k=5) -> list: ...
    def remove(self, mem_id): ...

class VectorMemoryRepo(MemoryRepo):
    def __init__(self, index, tenant, user):
        self.index = index
        self.ns = f'{tenant}/{user}'  # scope
    def search(self, q, k=5):
        hits = self.index.query(embed(q),
            top_k=k, namespace=self.ns)
        return [Memory.from_hit(h)
                for h in hits]

Be carefulDo not let store-specific query syntax, such as filter dicts or bits of SQL, leak out through the repository. Once callers depend on it, the abstraction is gone.

Wrap one storage system's raw API in a small object with plain append, load and delete calls for one record type.

A TranscriptDAO gives the harness a few calls: append(session_id, message), load(session_id), list_sessions(user_id) and delete(session_id). A transcript is the saved record of a conversation. Inside the DAO sit the SQL, the connection pool (a set of open database connections it reuses) and the JSON encoding of content blocks. It also handles paging through long transcripts. The session service and the memory repository call the DAO, not the database driver. So moving transcripts from a database file on a laptop to a managed SQL server in production changes only one class.

class TranscriptDAO:
    def __init__(self, db): self.db = db
    def append(self, sid, msg):
        self.db.execute(
          'INSERT INTO turns VALUES (?,?,?)',
          (sid, msg.seq, msg.to_json()))
    def load(self, sid):
        rows = self.db.query(
          'SELECT body FROM turns WHERE sid=?'
          ' ORDER BY seq', (sid,))
        return [Message.from_json(r)
                for r in rows]

Be carefulA DAO that just copies SQL one-to-one adds little. Name its methods in harness terms, such as append a turn or load a session. Give each record type its own DAO.

Choose a storage system once at startup and get a matching set of DAOs (transcripts, memories, artifacts) that all use it.

The harness reads its config at startup. Local mode wants an embedded database (one stored in a local file) and the local file system. Cloud mode wants a managed SQL server and object storage (a cloud service for storing files). Evals (automated test runs) want everything in memory. There is one factory per mode. It returns transcripts(), memories() and artifacts(), all connected to the same storage and credentials. A DAO is a small object that handles storage for one kind of record. The session service asks the factory for DAOs and never checks the mode itself. So a laptop run, a hosted run and an eval run can never mix stores.

class DAOFactory(Protocol):
    def transcripts(self) -> TranscriptDAO:...
    def artifacts(self) -> ArtifactDAO: ...

class LocalFactory(DAOFactory):
    def transcripts(self):
        return FileTranscriptDAO(self.root)
    def artifacts(self):
        return FileArtifactDAO(self.root)

FACTORIES = {'local': LocalFactory,
             'cloud': CloudFactory,
             'eval': MemoryFactory}
f = FACTORIES[cfg.mode](cfg)   # once

Be carefulIf you will only ever have one storage system, the factory is just extra ceremony. Add it when a second real one arrives. In-memory fakes for evals count.

Keep objects like Memory and Turn unaware of storage by putting a separate mapper between them and the store.

In the harness, a Memory has text, source, created_at and importance. It has methods like is_stale(). The vector store (a database that finds text by meaning) wants something else. It wants an id, a list of numbers called a vector, and a flat set of text labels called metadata. A MemoryMapper converts between the two. One way is to_record: it embeds the text (turns it into a vector), flattens the fields and turns dates into text. The other way is from_hit: it reads the metadata and adds the similarity score (how close the match was). The Memory class imports nothing from the store's client library. So you can test it without a database. You can also move it to a different store by rewriting only the mapper.

class MemoryMapper:
    def to_record(self, m: Memory):
        return {'id': m.id,
            'vector': embed(m.text),
            'meta': {'tenant': m.tenant,
                'text': m.text, 'src': m.source,
                'ts': m.ts.isoformat()}}
    def from_hit(self, h) -> Memory:
        md = h.meta
        return Memory(id=h.id,
            tenant=md['tenant'],
            text=md['text'], source=md['src'],
            ts=parse_iso(md['ts']),
            score=h.score)

Be carefulA mapper roughly doubles the code compared with objects that save themselves. For a few flat records with no logic, that cost may not be worth it.

Send flat, versioned data objects with no behaviour between separate processes, such as the tool runner and the chat client.

Sometimes the agent loop talks to a tool runner in another process, inside a sandbox (a locked-down space for running tools). Sometimes the server streams events to the chat client. In both cases it sends DTOs: ToolRequest {call_id, name, input}, ToolResult {call_id, content, is_error} and StreamEvent {type, delta}. They are frozen dataclasses, which means they cannot change after they are made. They turn into JSON and carry a schema version number. Live objects such as a Session or a ToolCommand hold open files and methods. Those never cross the wire. At the boundary, the harness copies only the needed fields into a DTO.

@dataclass(frozen=True)
class ToolRequest:
    v: int          # schema version
    call_id: str
    name: str
    input: dict

req = ToolRequest(1, b.id, b.name, b.input)
r = http.post(RUNNER, json=asdict(req))
res = ToolResult(**r.json())

Be carefulDo not let DTOs grow methods or hold live handles such as open files or clients. Give them version numbers, because the two sides of the wire are updated at different times.

Treat content blocks, messages and token counts as values that never change and are compared by content, so sharing and caching them is safe.

TextBlock('hello'), a ToolUse block and TokenCount(input=1200, output=340) are value objects. They are frozen (they cannot change), and equality and hashing use their fields. A hash is a short fingerprint made from the content. The harness relies on this in several places. Removing duplicates compares blocks by value. Local caches use a hash of block values as the key. Usage totals add TokenCount values to make new ones, instead of changing a counter that two turns share. The provider has a prompt cache (saved work it can reuse when a request starts the same way). It matches the exact bytes of the start of the request. So the harness needs blocks that never change plus deterministic serialisation (turning data into text the same way every time). Together they keep a resent prefix byte-identical. A block in a prefix that was already sent must never change in place. If it does, the cached prefix silently stops matching.

@dataclass(frozen=True)
class TokenCount:
    input: int = 0
    output: int = 0
    def __add__(self, o):
        return TokenCount(
            self.input + o.input,
            self.output + o.output)

a, b = TextBlock('hello'), TextBlock('hello')
assert a == b and hash(a) == hash(b)
total = total + turn.usage   # new object

Be carefulA frozen dataclass that holds a list can still be changed through that list. Use tuples (lists that cannot change) inside value objects, and never give them ids.

Load and save a whole session (messages, todos, tool state, files read) as one unit through a single root object.

A Session object is the root of an aggregate (a group of objects that are always handled together). The group holds messages, a todo list and tool state, such as the shell's current folder. It also holds the set of files the agent has read and the permission grants for this session. The harness loads it with sessions.get(id) and saves it with sessions.save(s). Only the root has a public id. Code reaches the parts through the root, as in s.todos.add or s.messages.append. So resuming or forking a session moves one consistent snapshot, not five loosely linked rows. A subagent is a helper agent started by the main one. When a subagent gets a copy, the harness removes the grants and gives it a smaller set. Approvals belong to the person or agent who gave them.

class Session:              # aggregate root
    def __init__(self, sid):
        self.id = sid
        self.messages: list[Message] = []
        self.todos = TodoList()
        self.tool_state = {'cwd': '/repo'}
        self.read_files: set[str] = set()

s = sessions.get(sid)
s.todos.add('run tests')
sessions.save(s)    # one write, whole graph

Be carefulIf every save rewrites megabytes of transcript, the group is too big. Keep fast-growing parts like messages in their own log, and keep only references to them in the root.

Collect a turn's memory, todo and message writes, then save them all together at the end of the turn, or cancel them all.

During one agent turn, the model might save two memories, update the todo list, write an artifact and add three messages. The harness does not write each change right away. It registers each one with a TurnUnitOfWork. If the turn ends cleanly, uow.commit() writes everything in one transaction (a group of database writes that all succeed or all fail). The turn might fail instead. The user might cancel, a token limit might be hit, or a tool might throw an error. Then uow.rollback() drops the waiting writes. The saved session never shows half a turn. A retried turn does not create duplicate memories.

def run_turn(session, uow):
    reply = call_model(session)
    for m in reply.new_memories:
        uow.register_new(m)
    uow.register_dirty(session.todos)
    uow.register_new(reply.message)

with TurnUnitOfWork(db) as uow:
    run_turn(session, uow)
# __exit__: commit() on success,
# rollback() on any exception

Be carefulRollback cannot undo a sent email or a git push. It also must not erase the record of a tool that already ran. Log executed tool calls outside the unit of work. Put a separate check in front of tools that cannot be undone.

Keep one in-memory object per document id while building context, so a file or memory found twice enters the window once.

In one turn, vector search (search by meaning) returns chunk doc-17#3, a piece of a document. A grep tool (a text search tool) finds the same part of the file. A memory lookup pulls it in again. Without care, the context window (the text the model can see at one time) holds three copies. That wastes tokens and suggests to the model that the fact is extra important. The harness keeps an identity map for each turn. Its key is the source and chunk id, or the path and a content hash (a fingerprint of the content). It returns the object that is already loaded. The context builder then cites it as [2] instead of pasting it again. If a tool edits a file, the file is also updated in only one place.

class IdentityMap:
    def __init__(self): self._m = {}
    def get(self, key, load):
        if key not in self._m:
            self._m[key] = load(key)
        return self._m[key]

idmap = IdentityMap()          # per turn
for hit in rag_hits + grep_hits:
    doc = idmap.get(hit.key, store.load)
    ctx.cite(doc)      # same doc, same [n]

Be carefulLimit the map to one turn or one session. A map that lives forever across the whole program keeps serving old file contents after the file changes on disk.

Put short placeholders for tool schemas and file contents into context, and load the full version only when the agent asks.

Say a harness is connected to 200 tools through MCP (a standard way to connect tools to an AI model). It does not paste every tool's JSON schema (the description of a tool's inputs) into the request. It calls tools/list once per server, which returns every tool's full definition. It keeps those definitions in a local registry (a lookup list in the harness). It sends the model only a tool search tool plus short descriptions of what is available. MCP has no way to fetch one tool's schema on its own. So the lazy step is putting a stored schema into context, not fetching it from the server. When the model searches for a tool, the harness loads that one full schema into context. Then the model can call it with valid arguments. The harness adds discovered schemas after the cached prefix, as tool search on the provider side does. That keeps the prompt cache (saved work the provider can reuse when a request starts the same way) valid. Editing the front of the tools list would break it from that point on. Files work the same way. Context holds a FileRef with the path, size and a summary. The content is read only when the model calls read_file.

class LazyTool:
    def __init__(self, name, blurb, reg):
        self.name, self.blurb = name, blurb
        self._reg, self._schema = reg, None
    @property
    def schema(self):
        if self._schema is None:  # first use
            self._schema = self._reg.load(
                self.name)
        return self._schema

# reg was filled once from tools/list
stub = {'name': t.name, 'desc': t.blurb}

Be carefulEach lazy load costs a round trip. A loop over lazy references turns into N separate fetches. Load ahead of time the few tools or files that almost every turn uses.

Share one unchangeable copy of the system prompt and tool schemas across thousands of sessions, keeping only each session's own state separate.

A hosted harness may run 10,000 sessions with the same 8,000-token system prompt and 40 tool schemas. A flyweight factory stores that bundle once, keyed by a content hash (a fingerprint of its contents). Every session holds a reference to one frozen PromptBundle that cannot change. Each session stores only its own messages, user id and settings. Sharing one object in memory does not by itself give cache hits. But every session turns that bundle into the same bytes. So requests also share a prompt-cache prefix at the API. The prompt cache is saved work the provider can reuse when a request starts the same way. This works as long as the requests fall within the cache's lifetime and scope. Details for each session, such as the user's name or today's date, go after the shared prefix. They are the extrinsic state.

_bundles: dict[str, PromptBundle] = {}

def bundle_for(version, tools):
    key = hash_of(version, tools)
    if key not in _bundles:
        _bundles[key] = PromptBundle.build(
            version, tools)       # frozen
    return _bundles[key]

s = Session(user, bundle=bundle_for(v, t))

Be carefulNever put details about a user into the shared bundle. A name leaked into the flyweight shows up in every session and breaks the cache prefix.

Write each tool call to a lasting log before running it, so a crashed run can restart exactly where it stopped.

Before the harness runs a tool call, it adds a tool_use record to the session's JSONL transcript. JSONL is a file with one JSON record per line. Then it calls fsync, which forces the data onto the disk. When the tool returns, it adds the tool_result. If the process dies in the middle of a turn, the harness reads the end of the log on restart. A tool_use with no matching tool_result marks an interrupted call. The harness reruns the call if the tool is idempotent (safe to run twice with the same result). Otherwise it tells the model the call was interrupted. It never silently loses the call or runs it twice.

def run_tool(call):
    wal.append({'type': 'tool_use',
                'id': call.id,
                'in': call.input})
    wal.fsync()              # durable first
    out = tools[call.name](**call.input)
    wal.append({'type': 'tool_result',
                'id': call.id, 'out': out})
    wal.fsync()

def recover():
    return [e for e in wal.tail()
            if unmatched(e)]

Be carefulFlushing to disk on every streamed token is far too slow. Log at message and tool boundaries. And remember that the log does not make a tool that is unsafe to repeat safe to rerun.

Store a session as events that are only ever added, and rebuild any view (model context, chat UI, a fork) by replaying them.

The harness stores events such as user_message, assistant_message, tool_use, tool_result, compaction and rewind. Compaction means replacing old messages with a summary. Rewind means going back to an earlier point. The harness never stores a current context that it edits. The context sent to the model is a projection (a view built from the events). To build it, the harness replays the events, applies compaction summaries and drops branches that were rewound. The chat UI is another projection, and a cost report is a third. Rewinding to a checkpoint, or forking a subagent (a helper agent) from turn 12, means replaying events up to event 12 into a new stream. Nothing in the original history is overwritten.

events = store.read(session_id)

def project_context(events, upto=None):
    msgs = []
    for e in events[:upto]:
        if e.type == 'message':
            msgs.append(e.msg)
        elif e.type == 'compaction':
            msgs = [e.summary]
        elif e.type == 'rewind':
            msgs = msgs[:e.to]
    return msgs

Be carefulReplaying thousands of events every turn gets slow. Save snapshots of the built context from time to time, and replay only the events after the latest snapshot.

Mark a memory or summary as changed when its source changes, and re-embed or re-summarise only when that mark is set.

Re-embedding a memory file (turning its text into numbers for search again) costs model calls. So does making a new conversation summary. The harness keeps a dirty flag, or a saved content hash (a fingerprint of the content), for each derived item. A derived item is one built from other data. When someone edits a memory note, the file watcher sets embedding_dirty. When new messages pass the summary point, it sets summary_dirty. Before each turn, a refresh step recomputes only the items whose flag is set, then clears the flag. A quiet session with no new content costs nothing extra. The cached prompt prefix also stays byte-identical, so the provider can keep reusing it.

class MemoryNote:
    def __init__(self, text):
        self.text, self.vec = text, None
        self.dirty = True
    def edit(self, text):
        self.text, self.dirty = text, True
    def vector(self):
        if self.dirty:        # changed?
            self.vec = embed(self.text)
            self.dirty = False
        return self.vec

Be carefulA missed flag is worse than a wasted recompute. If any write path forgets to set it, the agent works from an old summary. When writes come from many places, use content hashes instead.

Save a whole live object, such as an agent checkpoint, as one blob of data that can be stored, sent and restored elsewhere.

Sometimes a long-running agent pauses to wait for approval. Sometimes a scheduled task hands work to a fresh worker (another machine or process). In both cases the harness serialises a Checkpoint. It holds the messages, todo list, tool state, the waiting tool call, model settings and a schema version number. The blob (one block of stored data) goes to object storage (a cloud service for files) or into a queue message. Any worker can deserialise it and carry on the loop. So a run can wait hours for a human without keeping a live process. Moving a session between machines is just a file copy.

@dataclass
class Checkpoint:
    v: int
    messages: list
    todos: list
    pending_call: dict | None

blob = gzip(json.dumps(asdict(ckpt)))
blobs.put(f'ckpt/{sid}', blob)
# later, on any worker:
d = json.loads(gunzip(blobs.get(key)))
ckpt = Checkpoint(**migrate(d))

Be carefulNever unpickle checkpoints from sources you do not trust. Unpickling (loading Python's pickle format) can run hidden code. Keep a version field and upgrade steps, because yesterday's blob must still load in today's code.

Store a transcript or artifact body as one packed blob column, in a row whose other columns stay small and searchable.

The sessions table has normal columns: id, user_id, title, model, updated_at and token_total. It also has a body column. That column holds the compressed JSON of every message and content block. Listing a user's recent sessions, or adding up tokens per day, reads only the small columns. The harness reads the full blob only when someone opens the session. It never needs a separate table for each kind of content block (text, tool_use, image, citation).

db.execute(
    'INSERT INTO sessions(id, user_id,'
    ' title, tokens, updated, body)'
    ' VALUES (?,?,?,?,?,?)',
    (s.id, s.user, s.title, s.tokens,
     now(), gzip(json.dumps(s.messages))))

recent = db.query(
    'SELECT id, title FROM sessions'
    ' WHERE user_id=?'
    ' ORDER BY updated DESC LIMIT 20',
    (uid,))

Be carefulYou cannot search inside the blob efficiently, for example for every session that called the deploy tool. Copy any field you filter on into its own column or a side index.

Describe how memory records map to store fields in one table of settings, and let shared code do every load and save.

A harness may keep memories, artifacts and scheduled tasks in a vector store (a database that finds text by meaning) and in a SQL table. It does not write a separate mapper by hand for each type. Instead, each record class declares a mapping. For example, the text field goes to meta.body. user_id goes to meta.uid as a scope key (a field that limits whose data you can see). The vector is computed from the text by the embedder. One generic save() and one load() follow that map. Adding a field means adding a line. The same metadata can also generate the store's schema, and the optional filters that a memory search tool offers the model. Scope keys are different. The harness fills them from the caller's identity and never lets the model set them.

MEMORY_MAP = [
    Field('id',   to='id'),
    Field('text', to='meta.body'),
    Field('user', to='meta.uid', scope=True),
    Field('text', to='vector', via=embed),
]

def save(obj, mapping, store):
    rec = {}
    for f in mapping:
        val = f.convert(getattr(obj, f.name))
        put(rec, f.to, val)
    store.upsert(rec)

Be carefulMetadata hides behaviour. When a field quietly stops being saved, the bug is in a table nobody reads as code. So check the mapping when the program starts.

Save every kind of transcript event in one table with a kind column, leaving empty the fields a kind does not use.

A transcript holds many kinds of entry: UserMessage, AssistantMessage, ToolUse, ToolResult, Compaction and PermissionDecision. With single table inheritance, they all live in one events table. It has session_id, seq (the position in order) and kind. It also has columns that may be empty (NULL): text, tool_name, tool_input, is_error and summary. Loading a session is one SELECT, ordered by (session_id, seq). The loader picks the class from the kind column. Queries across every kind need no joins (combining tables). An example is everything in session 42 after position 300.

ROW_CLASS = {'user': UserMsg,
             'tool_use': ToolUse,
             'tool_result': ToolResult,
             'compaction': Compaction}

rows = db.query(
    'SELECT * FROM events WHERE sid=?'
    ' ORDER BY seq', (sid,))
events = [ROW_CLASS[r['kind']].from_row(r)
          for r in rows]

Be carefulEvery new kind adds mostly empty columns. The table cannot easily require tool_name only when kind is tool_use. Add CHECK constraints (database rules for each row), or switch to table inheritance when the kinds differ a lot.

Keep fields shared by all events in a base table, and each kind's own fields in a separate table linked by event id.

All agent events share id, session_id, seq (position in order), created_at and tokens. These live in a base events table. Each kind of event gets its own table with the same id. tool_calls has (id, tool, input, duration_ms, exit_code). approvals has (id, decision, reviewer). compactions has (id, summary, dropped_tokens). Timeline and cost queries read only the base table. Tool analytics join (combine) events with tool_calls. That table can have NOT NULL constraints (rules that a field must be filled) and an index on tool for fast lookups.

class Event(Base):        # table: events
    id: int; sid: str; seq: int; tokens: int
class ToolCall(Event):    # table: tool_calls
    tool: str; input: dict; exit_code: int
class Approval(Event):    # table: approvals
    decision: str; reviewer: str

slow = db.query(
    'SELECT e.sid, t.tool FROM events e'
    ' JOIN tool_calls t USING (id)'
    ' WHERE t.duration_ms > 30000')

Be carefulLoading a full mixed transcript needs a join, or one query per kind. So this layout suits audit and analytics better than the busy path of resuming a session.

Let the agent and a person edit one document without locks, and check at save time whether someone changed it first.