Build an Agent Harness From Scratch - the loop is thirty lines, the runtime is the rest

The last post argued that the model is not your agent - the harness around it is, and the harness is the part that lasts. That was the argument. This is the build.

We are going to write an agent harness in plain Python with the Anthropic SDK and nothing else. No framework. No abstraction we haven't earned. The rule for the whole post is simple: the loop stays small, and every concern is added as a function that hooks into it. By the end there are about 250 lines, and you will know exactly what every agent framework is doing on your behalf - which is the only honest position from which to decide whether to keep using one.

The order matters. Each step adds the thing that would have bitten you first if you shipped the previous step.

Step 0 - the loop

An agent is a while loop that calls a model, runs whatever tools the model asked for, feeds the results back, and repeats until the model stops asking. That is the whole thing. Everything else is what happens when you run it for more than five minutes.

# harness.py
import anthropic

client = anthropic.Anthropic()
MODEL = "claude-opus-5"


def run(goal: str, tools: list, system: str) -> str:
    messages = [{"role": "user", "content": goal}]

    while True:
        response = client.messages.create(
            model=MODEL,
            max_tokens=16000,
            system=system,
            tools=[t.schema for t in tools],
            messages=messages,
        )
        messages.append({"role": "assistant", "content": response.content})

        if response.stop_reason == "tool_use":
            results = [execute(t, block) for block in response.content if block.type == "tool_use"]
            messages.append({"role": "user", "content": results})
            continue

        if response.stop_reason == "pause_turn":
            continue  # the model paused mid-turn; calling again resumes it

        return text_of(response)


def text_of(response) -> str:
    return "".join(b.text for b in response.content if b.type == "text")

Three things in there are already load-bearing.

The transcript is the state. There is no other memory. Whatever the agent knows, it knows because it is in messages. Hold on to that - it becomes the persistence design in step 3.

stop_reason is the control flow. The model tells you why it stopped, and the harness decides what to do about it. We handle two reasons here. There are more - max_tokens, refusal - and we will come back for them.

Tool results go back as a user message. Not as text. Not as a summary. As tool_result blocks whose ids match the tool_use blocks that requested them. The model reads that message the same way it reads anything else you send it.

That loop will run. It will also read any file on your machine, delete anything it decides is in the way, and keep going until your credit card stops it. Let's fix that in order.

Step 1 - tools are a registry, not a pile of functions

A tool is three things: a schema the model sees, a handler the harness runs, and - the part most frameworks forget - a risk tier the permission layer will need in step 4. Put them in one place from the start.

from dataclasses import dataclass
from pathlib import Path
from typing import Callable, Literal

Risk = Literal["read", "write", "destructive", "spend"]


@dataclass
class Tool:
    name: str
    description: str
    input_schema: dict
    handler: Callable[..., str]
    risk: Risk = "read"

    @property
    def schema(self) -> dict:
        return {
            "name": self.name,
            "description": self.description,
            "input_schema": self.input_schema,
            "strict": True,  # the API guarantees the arguments match the schema
        }


def read_file(path: str) -> str:
    return Path(path).read_text()


def write_file(path: str, content: str) -> str:
    Path(path).write_text(content)
    return f"wrote {len(content)} bytes to {path}"


TOOLS = [
    Tool(
        name="read_file",
        description="Read a UTF-8 text file. Returns its full contents.",
        input_schema={
            "type": "object",
            "properties": {"path": {"type": "string"}},
            "required": ["path"],
            "additionalProperties": False,
        },
        handler=read_file,
        risk="read",
    ),
    Tool(
        name="write_file",
        description="Create or overwrite a UTF-8 text file with the given content.",
        input_schema={
            "type": "object",
            "properties": {"path": {"type": "string"}, "content": {"type": "string"}},
            "required": ["path", "content"],
            "additionalProperties": False,
        },
        handler=write_file,
        risk="write",
    ),
]

Now the executor. The one rule here is that a failed tool is still a result. If the handler throws, the model must hear about it, in the same message as every other result from that turn.

def execute(tool: Tool, block) -> dict:
    try:
        output = tool.handler(**block.input)
        return {"type": "tool_result", "tool_use_id": block.id, "content": output}
    except Exception as exc:
        return {
            "type": "tool_result",
            "tool_use_id": block.id,
            "content": f"{type(exc).__name__}: {exc}",
            "is_error": True,
        }

The descriptions are not decoration. They are the only documentation the model has. A vague description produces vague calls, and you will spend a week blaming the model for a problem you wrote.

Step 2 - the model behind a seam

Right now the loop calls client.messages.create directly. That is fine for a demo and wrong for anything you intend to keep, for the reason the previous post spent a section on: the model is the part most likely to change.

So the loop gets one function to talk to, and that function owns everything about talking to the model - retries, caching, and what happens when the model declines.

import time


def complete(system: str, tools: list, messages: list):
    for attempt in range(5):
        try:
            return client.beta.messages.create(
                model=MODEL,
                max_tokens=16000,
                system=system,
                tools=[t.schema for t in tools],
                messages=messages,
                cache_control={"type": "ephemeral"},  # cache the stable prefix
                betas=["server-side-fallback-2026-07-01"],
                fallbacks="default",  # if the model declines, re-run on a fallback server-side
            )
        except (anthropic.RateLimitError, anthropic.APIConnectionError) as exc:
            wait = 2**attempt
        except anthropic.APIStatusError as exc:
            if exc.status_code < 500:
                raise  # a 4xx is our bug, not the weather
            wait = 2**attempt
        emit("retry", attempt=attempt, wait=wait, error=type(exc).__name__)
        time.sleep(wait)
    raise RuntimeError("model unavailable after 5 attempts")

Three decisions are hiding in there.

Retries are the adapter's job. The SDK already retries twice on its own; this wraps that with a longer, logged backoff, because an agent mid-task should survive a bad minute at the provider. A 4xx is different - that is a malformed request, and retrying it is just paying to see the same error.

Caching is the adapter's job. The system prompt and tool list are identical on every iteration of the loop. That is exactly the shape prompt caching is for. The one-line cache_control tells the API to cache the stable prefix, and on a long run most of your input tokens come from cache at a tenth of the price. Check response.usage.cache_read_input_tokens - if it is zero after the second call, something in your prefix is changing between calls, and it is usually a timestamp.

Declines are the adapter's job. A safety classifier can end a turn with stop_reason: "refusal". The fallbacks parameter re-runs the request on a fallback model inside the same call, so the loop only sees a refusal when the whole chain declined. That is a case worth stopping on, and the final loop does.

Step 3 - the transcript is the state, so make it durable

The loop's messages list is the entire memory of the run. If the process dies at step 40, all of it is gone and you start again from the goal, repeating every tool call - including the ones with side effects.

The fix is almost embarrassingly small: write every message to disk as it is appended, and reload on start.

import json


class Transcript:
    def __init__(self, path: Path):
        self.path = path
        self.messages = []
        if path.exists():
            self.messages = [json.loads(line) for line in path.read_text().splitlines()]

    def append(self, role: str, content) -> None:
        if isinstance(content, list):
            content = [b.model_dump() if hasattr(b, "model_dump") else b for b in content]
        message = {"role": role, "content": content}
        self.messages.append(message)
        with self.path.open("a") as f:
            f.write(json.dumps(message) + "\n")

One file. One line per message. Append-only. That gives you three things for free:

• Resume. Restart the process with the same path and the next model call continues from wherever the last one left off.
• Audit. The file is the record of what the agent saw and did. There is no second logging system to keep in sync with it.
• Replay. Point a different model at the same transcript and compare the decision it makes. That is your evaluation harness, and you built it by accident.

While we are here: budgets. A loop that runs until the model stops is a loop that runs until your money does.

@dataclass
class Budget:
    max_steps: int = 50
    max_seconds: float = 900.0
    max_input_tokens: int = 2_000_000

    def check(self, steps: int, elapsed: float, input_tokens: int) -> None:
        if steps >= self.max_steps:
            raise BudgetExceeded(f"{steps} steps")
        if elapsed >= self.max_seconds:
            raise BudgetExceeded(f"{elapsed:.0f}s")
        if input_tokens >= self.max_input_tokens:
            raise BudgetExceeded(f"{input_tokens} input tokens")


class BudgetExceeded(RuntimeError):
    pass

Steps, seconds and tokens. Three different ways a run can go wrong, and none of them is "the model produced a bad answer" - they are all "the harness let it keep going". Tokens are the interesting one: sum response.usage.input_tokens across calls and you are watching the cost in real time.

Step 4 - permissions, or the tool call that never ran

Step 1 gave every tool a risk tier. Now use it. Between "the model asked for a tool" and "the handler ran" there is a gate, and the gate answers one question: allow, ask, or deny?

Decision = Literal["allow", "ask", "deny"]
WORKSPACE = Path.cwd().resolve()


def policy(tool: Tool, args: dict) -> Decision:
    if tool.risk == "read":
        return "allow"
    if tool.risk == "write":
        target = (WORKSPACE / args.get("path", "")).resolve()
        return "allow" if target.is_relative_to(WORKSPACE) else "deny"
    return "ask"  # destructive and spend always stop for a human


def gate(tool: Tool, block, approve: Callable[[Tool, dict], bool]) -> dict | None:
    decision = policy(tool, block.input)
    if decision == "allow":
        return None
    if decision == "ask" and approve(tool, block.input):
        emit("approved", tool=tool.name, args=block.input)
        return None
    return {
        "type": "tool_result",
        "tool_use_id": block.id,
        "content": f"Permission denied: {tool.name} is a {tool.risk} operation and was not approved.",
        "is_error": True,
    }

gate returns either nothing - go ahead and run it - or a finished tool_result that says no. The loop doesn't care which; it just uses the gate's result if there is one and calls execute otherwise.

results.append(gate(tool, block, approve) or execute(tool, block))

Notice what the denial looks like. It is a tool result, marked as an error, with a sentence explaining why. The model always hears the answer. The tempting shortcut is to skip the call and say nothing, and it is wrong: from the model's side a silently dropped call is indistinguishable from a tool that hung, and it will retry, or route around you, or both.

approve is injected. In a terminal it is input(). In a service it posts to a queue and waits. In a test it is lambda *_: False. The harness doesn't know and shouldn't.

Step 5 - context is a budget too

Every iteration re-sends the whole transcript. On a long run the transcript is dominated by tool results the model needed once - the 4,000-line file it read at step 3 and never looked at again. That costs money on every subsequent call and, worse, crowds out the recent turns that actually matter.

The simplest fix is to make the model call for it again if it needs it. Ask the API to clear old tool results out of the context it sees, while your transcript on disk keeps every byte:

context_management={"edits": [{"type": "clear_tool_uses_20250919"}]},
betas=[..., "context-management-2025-06-27"],

Add those two lines to complete() and the server drops stale tool results from what the model reads, oldest first, when the context gets large. Your transcript is untouched, which is exactly the split you want: the record is complete, the working set is small.

You could do this client-side by rewriting old tool_result blocks to a stub before each call, and older harnesses did. Don't. That is exactly the history edit the warning in step 3 was about. The server-side version exists so that the model's own reasoning stays consistent with what it can see.

For runs that outgrow even that, the same parameter has a compaction mode that summarises rather than clears. Start with clearing. It is cheaper, it is lossless on disk, and most runs never need more.

Step 6 - observability, or one line per decision

You will be asked "why did it do that?" and the transcript answers "what did it see?", which is close but not the same. The harness made decisions the transcript doesn't show: it retried, it denied, it trimmed, it ran out of budget. Those need a record of their own.

import sys
import uuid

RUN_ID = uuid.uuid4().hex[:8]


def emit(event: str, **fields) -> None:
    record = {"ts": round(time.time(), 3), "run": RUN_ID, "event": event, **fields}
    print(json.dumps(record), file=sys.stderr)

That's it. Structured, one line, timestamped, tagged with the run. Swap print for your tracing library when you have one; the shape is what matters. Emit from every place the harness makes a choice:

The model call is the event worth being greedy about. Log stop_reason, usage.input_tokens, usage.output_tokens and usage.cache_read_input_tokens every time, and you have a cost-per-step chart before you have a dashboard.

Step 7 - recovery, or what to do when it's stuck

Failures inside a tool are already handled: they come back as is_error results and the model routes around them. Failures at the provider are handled by complete(). That leaves the failure mode that is specific to agents: the loop that isn't going anywhere.

The classic shape is the same tool with the same arguments, three times in a row. The model tried something, it didn't work, and it tried it again because nothing in its context told it not to.

def looping(recent: list, block, window: int = 3) -> bool:
    key = (block.name, json.dumps(block.input, sort_keys=True))
    return len(recent) >= window and all(k == key for k in recent[-window:])

Keep a short list of (name, args) for the last few calls. When the window fills with identical entries, don't run the tool - hand back an error result that says so:

LOOP_MESSAGE = (
    "Loop detected: this exact call has been made three times. "
    "Do not repeat it. Change approach, or report what is blocking you."
)


def error_result(block, message: str) -> dict:
    return {"type": "tool_result", "tool_use_id": block.id, "content": message, "is_error": True}


if looping(recent, block):
    results.append(error_result(block, LOOP_MESSAGE))
    continue

This is the same trick as the permission gate - the harness talks to the model through tool results, because that is a channel the model is already listening on. No special protocol, no prompt surgery, no second model watching the first.

The other recovery case is max_tokens. The model hit the output ceiling mid-thought. The transcript is still valid, so the answer is a nudge and another call:

if response.stop_reason == "max_tokens":
    transcript.append("user", "You hit the output limit. Continue from where you stopped.")
    continue

Bounded, as always, by the step budget.

Step 8 - done means verified, not stopped

end_turn means the model stopped. It does not mean the goal was met. Every serious agent harness has some notion of checking the result before declaring victory - a test run, a schema check, a second look - and it belongs in the loop, not after it.

@dataclass
class Verdict:
    ok: bool
    detail: str = ""


def verify_with_tests(goal: str, answer: str) -> Verdict:
    import subprocess
    proc = subprocess.run(["pytest", "-q"], capture_output=True, text=True)
    return Verdict(ok=proc.returncode == 0, detail=proc.stdout[-2000:])

The verifier is injected like approve. When it fails, the failure goes back to the model as the next user message and the loop continues. When it passes, the run ends. A verifier that is code - tests, a linter, a JSON schema - is the highest-leverage line in the harness, because it converts "the model said it's done" into "the model was checked".

The assembled loop

Here is the whole loop with every hook in place. Compare it to step 0: it is longer, but every extra line is a call into one of the functions above, and the shape has not changed.

def run(goal, tools, system, approve, verify, transcript, budget) -> str:
    by_name = {t.name: t for t in tools}
    if not transcript.messages:
        transcript.append("user", goal)

    started, steps, input_tokens, recent = time.monotonic(), 0, 0, []
    emit("run.start", goal=goal, resumed=len(transcript.messages) > 1)

    while True:
        budget.check(steps, time.monotonic() - started, input_tokens)
        steps += 1

        response = complete(system, tools, transcript.messages)
        input_tokens += response.usage.input_tokens
        emit("model", stop=response.stop_reason, usage=response.usage.model_dump())
        transcript.append("assistant", response.content)

        if response.stop_reason == "tool_use":
            results = []
            for block in (b for b in response.content if b.type == "tool_use"):
                tool = by_name.get(block.name)
                if tool is None:
                    results.append(error_result(block, f"unknown tool {block.name}"))
                elif looping(recent, block):
                    results.append(error_result(block, LOOP_MESSAGE))
                else:
                    recent.append((block.name, json.dumps(block.input, sort_keys=True)))
                    started_at = time.monotonic()
                    result = gate(tool, block, approve) or execute(tool, block)
                    emit("tool", name=tool.name, ms=round((time.monotonic() - started_at) * 1000),
                         error=result.get("is_error", False))
                    results.append(result)
            transcript.append("user", results)
            continue

        if response.stop_reason == "pause_turn":
            continue

        if response.stop_reason == "max_tokens":
            transcript.append("user", "You hit the output limit. Continue from where you stopped.")
            continue

        if response.stop_reason == "refusal":
            emit("refused", details=response.stop_details.model_dump() if response.stop_details else None)
            raise RuntimeError("the model declined the task")

        answer = text_of(response)
        verdict = verify(goal, answer)
        emit("verify", ok=verdict.ok)
        if verdict.ok:
            emit("run.end", steps=steps, input_tokens=input_tokens)
            return answer
        transcript.append("user", f"Verification failed:\n{verdict.detail}\nFix it and try again.")

Running it is one call:

run(
    goal="Make the failing test in tests/test_parser.py pass without changing the test.",
    tools=TOOLS,
    system="You are a careful engineer working in this repository. Read before you write.",
    approve=lambda tool, args: input(f"{tool.name}({args})? [y/N] ").lower() == "y",
    verify=verify_with_tests,
    transcript=Transcript(Path(".runs/parser-fix.jsonl")),
    budget=Budget(max_steps=40),
)

Kill it halfway and run it again. It picks up from the transcript. Swap the verifier for one that checks a JSON schema and it becomes a data-extraction agent. Swap approve for a Slack message and it becomes something a team can supervise. None of those changes touch the loop.

What was deliberately left out

Some things a production harness needs are missing here, and it is worth being explicit about which and why.

Each is a real concern. None of them changes the loop.

So should you actually roll your own?

Probably not for production - and that is not the point of having done it.

Anthropic's SDK ships a tool runner that does steps 0 and 1 for you with per-turn hooks for the rest. The Agent SDK ships the whole thing with built-in tools. Managed Agents ships it and hosts it. Every one of those is a better default than 250 lines you maintain alone.

But you cannot evaluate a harness you don't understand. When the framework's permission model doesn't fit your compliance rule, when its context strategy throws away the thing your agent needed, when its retry policy hides a 4xx you should have seen - you need to know which of the eight steps above it got wrong, and whether its hooks let you fix it. The build was for that.

The model provides intelligence. The loop provides the system. Now you have seen how little the loop is, and how much the system is.

This is the hands-on companion to The Model Is Not Your Agent, which makes the case for why the harness matters. For what sits around the harness: agent containment covers everything the permission gate cannot enforce, MCP and A2A cover the boundaries between agents, and the AI war moving down the stack covers why the runtime, not the model, is where the moat is.