The Agent Loop: Plan, Act, Observe, Repeat

The agent loop is usually drawn as four boxes in a circle. The boxes hide the only interesting part, which is what exactly gets appended to the message list and in what order. Here is the whole thing as code.

The shape

Send the conversation. If the reply contains tool calls, execute them, append their results, and send the conversation again. If it does not, you have an answer. That is the entire control structure — the loop body is a single branch, and every framework in existence is wrapping those ten lines in configuration.

What makes it non-trivial is that the conversation is append-only and the API is stateless. Every iteration re-sends everything: system prompt, tool schemas, every previous assistant message and every tool result. Step twelve is not one API call, it is the twelfth increasingly expensive API call, which is the fact behind almost everything in agent cost control.

Eighty lines, no framework

Against any OpenAI-compatible chat completions endpoint. Two real tools, a step cap, error capture, and output truncation.

import json, os, pathlib, requests

API   = "https://your-gateway.example/v1/chat/completions"
MODEL = "some-capable-model"
ROOT  = pathlib.Path("./workspace").resolve()
HEAD  = {"Authorization": "Bearer " + os.environ["API_KEY"]}

SYSTEM = (
    "You inspect a small codebase. Use list_dir and read_file to find "
    "evidence before answering. When you can answer, answer in prose "
    "and cite the file paths you read. Do not guess at file contents."
)

TOOLS = [
    {"type": "function", "function": {
        "name": "list_dir",
        "description": "List file and directory names directly inside a "
                       "directory, relative to the workspace root. Not "
                       "recursive. Use this before read_file when you do "
                       "not already know a path exists.",
        "parameters": {"type": "object", "properties": {
            "path": {"type": "string",
                     "description": "Relative dir, '.' for the root."}},
            "required": ["path"]}}},
    {"type": "function", "function": {
        "name": "read_file",
        "description": "Return the UTF-8 text of one file, truncated to "
                       "8000 characters. Fails if the path does not exist "
                       "or escapes the workspace root.",
        "parameters": {"type": "object", "properties": {
            "path": {"type": "string",
                     "description": "Relative file path from list_dir."}},
            "required": ["path"]}}},
]

def _safe(rel: str) -> pathlib.Path:
    p = (ROOT / rel).resolve()
    if not str(p).startswith(str(ROOT)):     # blocks ../../etc/passwd
        raise ValueError("path escapes the workspace root")
    return p

def list_dir(path="."):
    return "\n".join(sorted(q.name + ("/" if q.is_dir() else "")
                            for q in _safe(path).iterdir()))

def read_file(path):
    return _safe(path).read_text("utf-8")[:8000]

IMPL = {"list_dir": list_dir, "read_file": read_file}

def call_model(messages):
    r = requests.post(API, headers=HEAD, timeout=120, json={
        "model": MODEL, "messages": messages,
        "tools": TOOLS, "tool_choice": "auto"})
    r.raise_for_status()
    body = r.json()
    return body["choices"][0]["message"], body.get("usage", {})

def run(task, max_steps=12):
    messages = [{"role": "system", "content": SYSTEM},
                {"role": "user",   "content": task}]
    spent_in = spent_out = 0

    for step in range(max_steps):
        msg, usage = call_model(messages)
        spent_in  += usage.get("prompt_tokens", 0)
        spent_out += usage.get("completion_tokens", 0)
        messages.append(msg)                  # verbatim, tool_calls included

        calls = msg.get("tool_calls") or []
        if not calls:
            return {"answer": msg.get("content"), "steps": step + 1,
                    "in": spent_in, "out": spent_out, "stop": "model_done"}

        for call in calls:                    # every id gets a reply
            name = call["function"]["name"]
            try:
                args = json.loads(call["function"]["arguments"])
                out  = str(IMPL[name](**args))
            except Exception as e:
                out  = "ERROR " + type(e).__name__ + ": " + str(e)
            messages.append({"role": "tool", "tool_call_id": call["id"],
                             "name": name, "content": out[:8000]})

    return {"answer": None, "steps": max_steps, "in": spent_in,
            "out": spent_out, "stop": "step_budget_exhausted"}

if __name__ == "__main__":
    print(run("Which file defines the retry policy, and what is the cap?"))

What each part is doing

• messages.append(msg) puts the assistant’s reply back unmodified, including its tool_calls array. This is not optional bookkeeping; see the first bug below.
• The inner for call in calls loop runs sequentially and appends one role: “tool” message per call. Running them concurrently is legal and sometimes wrong — that is a page of its own.
• The try/except turns a tool crash into a string the model reads. This is the single highest-leverage line in the file: a model that is told “ERROR FileNotFoundError: no such file” will usually call list_dir and recover, whereas an exception that propagates kills the run.
• out[:8000] exists because one cat of a lockfile can consume the context window, and the loop will then fail on every subsequent call for reasons that look unrelated.
• range(max_steps) rather than while True, and the return value carries a stop reason. A loop that cannot say why it ended is a loop you cannot alert on.
• _safe() is here because read_file takes a path from a stochastic process. Path traversal is not an exotic attack; the model will attempt ../ by accident within a week.
• timeout=120 on the HTTP call is not boilerplate. Without it a hung connection stalls the run indefinitely, and the wall-clock halt is then the only thing that will ever notice.

The system prompt is doing more work than its three lines suggest. “Use list_dir and read_file to find evidence before answering” is a policy about tool ordering, which no schema can express. “Do not guess at file contents” is there because the failure it prevents — a confident summary of a file that was never opened — passes every structural check the loop performs. The same loop with a two-line system prompt behaves noticeably worse, and the paragraph costs about eighty tokens per step.

Four bugs everyone writes once

Dropping the assistant message

It is tempting to append only the tool results, since the assistant message “had no content”. The API rejects it. OpenAI returns a 400 whose message is, verbatim: Invalid parameter: messages with role ‘tool’ must be a response to a preceding message with ‘tool_calls’. The tool result is meaningless without the call it answers, so both go in the history, in order.

Answering only some of the tool calls

When a model emits three calls in one message, you owe three tool messages before the next request. Skip one — because it errored, or because you short-circuited on the first useful result — and the next call fails validation for the same structural reason. Errors are still answers; send them.

Treating arguments as an object

It is a JSON string, and it is generated text, so it can be malformed. json.loads belongs inside the try. A JSONDecodeError handed back to the model as text is usually fixed on the next step; the same exception unhandled ends the run.

Assuming a text reply means success

The loop above returns on the first message with no tool calls. That message might be “I was unable to find the file.” The loop reports model_done, and the caller reads it as an answer. The fix is an explicit termination handshake — a finish tool the model must call, with a structured result — covered in stopping conditions.

What this deliberately does not do

No streaming, no retries on 429 or 5xx, no cost cap, no compaction when the context fills, no concurrency, no persistence across a crash, no tracing. Each of those is genuinely worth adding and each has a page in this cluster. Two of them change the shape of the code rather than adding to it: retries belong inside call_model rather than around run, because retrying the run replays every tool side effect that already happened, and persistence means writing messages to durable storage after each append, so a crash at step 30 resumes instead of restarting. That second one is cheap here and is the feature that most often justifies reaching for a framework at all. The point of showing the eighty lines is that the additions are additions — the control flow underneath any agent framework you adopt is the loop above, and when a framework behaves strangely, this is the mental model you debug it against.