Bài này cũng có bản tiếng Việt: Agent Loop: 40 dòng code mà mọi AI agent đều được xây trên đó. This is part 1 of a six-part series on AI agents in production. Part 2 is Agent Harness.

The first agent I put in front of real users was a document-processing bot for an operations team. It read scanned contracts, pulled out dates and parties, cross-checked them against a database and drafted a summary. The demo took two weeks. Getting it to behave for three months took the rest of the year, and almost none of that time went into prompts. It went into the loop around the model: when to stop, what to do when a tool failed, how much it was allowed to spend, and how to make a run reproducible when someone asked "why did it do that?"

That loop is the subject of this post. If you understand it well, every framework (LangGraph, the OpenAI Agents SDK, Claude Agent SDK, CrewAI, whatever ships next quarter) becomes a set of conveniences on top of something you could write yourself in an afternoon. If you don't understand it, the frameworks hide exactly the parts that will hurt you.

What an agent actually is#

A chat completion is one call: prompt in, text out. An agent is a model that can ask for work to be done, see the result, and decide what to do next. The whole trick is a loop:

1. Send the conversation to the model, together with a list of tools it may call.
2. If the model answers with plain text, you are done.
3. If the model answers with a tool call, run the tool, append the result to the conversation, and go to step 1.

That is it. Anthropic's "Building effective agents" post from December 2024 makes the same point in a more careful way: most successful systems are simple, composable patterns, and a fully autonomous agent is just the loop above with the model deciding how many times to go around. The ReAct paper from 2022 described the same shape before any of the current tooling existed.

What makes it an agent rather than a workflow is who controls the control flow. In a workflow, your code decides the sequence and the model fills in the blanks. In an agent, the model decides the sequence. That single difference is why agents can handle open-ended tasks, and also why they can spend forty dollars running ls in a loop.

When you should not build one#

Before the code, a filter I apply to every "let's make it an agent" request, because the loop is seductive and most tasks do not need it.

If you can write the sequence of steps on a whiteboard, write them in code and call the model inside each step. That is a workflow: cheaper, faster, testable with ordinary unit tests, and it fails in predictable places. Our contract pipeline eventually became exactly this. Extract, validate, cross-check, summarise, four model calls in a fixed order, with the agent loop kept only for the 5% of documents that failed validation and needed investigation.

Reach for the loop when the path to the answer depends on what you find along the way: debugging, research across many files, data cleaning where every dataset is broken differently, anything where a human expert would say "let me poke around first". The cost of an agent is variance. Runs take different amounts of time, money and steps, and some of them go wrong in ways you did not anticipate. You accept that variance in exchange for coverage of cases you could not enumerate. If you could enumerate them, you did not need the variance.

The honest middle ground is a workflow with agent steps inside it: a fixed outer skeleton, and a bounded loop at the one or two points where exploration is actually required. Most of what we ship looks like that.

The loop in 40 lines#

Here is the smallest version I would still call an agent. It uses the Anthropic SDK directly, no framework. You need pip install anthropic and an ANTHROPIC_API_KEY.

# agent_loop.py
import json
import os
import subprocess
from pathlib import Path

import anthropic

MODEL = os.environ.get("ANTHROPIC_MODEL", "claude-sonnet-5-5")
WORKDIR = Path(os.environ.get("AGENT_WORKDIR", ".")).resolve()
client = anthropic.Anthropic()

SYSTEM = (
    "You are a careful engineer working inside a sandboxed repository. "
    "Use tools to inspect files and run commands. Stop and answer when the task is done."
)

TOOLS = [
    {
        "name": "read_file",
        "description": "Read a UTF-8 text file relative to the working directory.",
        "input_schema": {
            "type": "object",
            "properties": {"path": {"type": "string"}},
            "required": ["path"],
        },
    },
    {
        "name": "run",
        "description": "Run a shell command in the working directory. Returns stdout, stderr and exit code.",
        "input_schema": {
            "type": "object",
            "properties": {"command": {"type": "string"}},
            "required": ["command"],
        },
    },
]


def read_file(path: str) -> str:
    target = (WORKDIR / path).resolve()
    if WORKDIR not in target.parents and target != WORKDIR:
        return "error: path escapes the working directory"
    return target.read_text(encoding="utf-8")


def run(command: str) -> str:
    proc = subprocess.run(
        command, shell=True, cwd=WORKDIR, capture_output=True, text=True, timeout=60
    )
    return json.dumps({"exit": proc.returncode, "stdout": proc.stdout, "stderr": proc.stderr})


def execute(name: str, args: dict) -> str:
    try:
        if name == "read_file":
            return read_file(**args)
        if name == "run":
            return run(**args)
        return f"error: unknown tool {name}"
    except Exception as exc:  # the model needs to see failures, not crash on them
        return f"error: {type(exc).__name__}: {exc}"


def agent(task: str, max_steps: int = 20, token_budget: int = 300_000) -> str:
    messages = [{"role": "user", "content": task}]
    spent, step = 0, 0

    while step < max_steps:
        step += 1
        response = client.messages.create(
            model=MODEL, max_tokens=4096, system=SYSTEM, tools=TOOLS, messages=messages
        )
        spent += response.usage.input_tokens + response.usage.output_tokens
        messages.append({"role": "assistant", "content": response.content})

        if response.stop_reason != "tool_use":
            return "".join(b.text for b in response.content if b.type == "text")

        results = []
        for block in response.content:
            if block.type == "tool_use":
                print(f"[{step}] {block.name} {json.dumps(block.input)[:120]}")
                output = execute(block.name, block.input)
                results.append(
                    {"type": "tool_result", "tool_use_id": block.id, "content": output[:12_000]}
                )
        messages.append({"role": "user", "content": results})

        if spent > token_budget and step < max_steps:
            messages.append({"role": "user", "content": "Budget exhausted. Summarise what you found and stop."})
            max_steps = step + 1  # exactly one more model call, then we stop

    raise RuntimeError(f"gave up after {step} steps")


if __name__ == "__main__":
    import sys
    print(agent(" ".join(sys.argv[1:]) or "List the files here and tell me what this project does."))

Run it in any repository:

export ANTHROPIC_API_KEY=sk-ant-...
python agent_loop.py "Find the slowest test in this repo and explain why it is slow"

Read the agent() function once more, slowly, because every production agent you will ever touch is this function with more decoration. The model gets the conversation and the tool list. If it stops for any reason other than tool_use, we return its text. Otherwise we run every tool call in the response, package the outputs as tool_result blocks keyed by the call id, append them as a user turn, and go around again.

Two details are easy to miss and both matter. First, a single response can contain several tool calls; you have to answer all of them, in one user message, or the API rejects the next request. Second, the tool result is a string you control. What you put in that string is the only thing the model will ever know about what happened. That is the first lever of agent quality, and most teams never touch it.

Where the loop breaks#

The toy above works. It is also exactly the version that will embarrass you in front of a customer, for four reasons I have hit personally.

It does not know when to stop. Models are optimistic. Given a vague task, they will keep calling tools long after the useful work is done, or declare victory after the first partial result. The max_steps and token_budget in the code are not nice-to-haves. They are the difference between a bounded cost and an invoice. In practice I set the step limit per task type: a lookup agent gets 6, a coding agent gets 40 to 60, and anything above that gets split into sessions (more on that in the harness post).

Tool errors become retry storms. If run raises instead of returning a string, the loop dies and the user sees a stack trace. If it returns a bare "error", the model will retry the same command three or four times with tiny variations. The fix is in execute(): catch everything, and return an error message that tells the model what to do differently. "Command timed out after 60s; try a narrower command or add a limit" produces a different next step than "error".

Tool output floods the context. One cat of a 2 MB log file and the rest of your context window is gone, along with your token budget. The output[:12_000] truncation is crude but necessary. A better version keeps the head and the tail and says how many bytes were dropped. The best version gives the model tools that cannot produce unbounded output in the first place (grep with a limit, read_file with line ranges).

You cannot explain what it did. The print in the loop is a stand-in for the thing you really need: a persisted trace of every model call, tool call and result, with timestamps and token counts. When a user asks why the agent changed the wrong file, "the model decided to" is not an answer. The trace is. Build it on day one; retrofitting it after an incident is miserable.

The four controls I never ship without#

Over a few agents in production, these became non-negotiable. They map directly onto the four failures above.

A termination policy that is explicit, not emergent. The model stopping is one terminating condition. The others are a step limit, a token or dollar budget, a wall-clock limit, and a "done" signal that the model has to request deliberately (a finish tool with a summary argument works well and gives you a structured final result instead of free text).

Tool results that teach. Every error string is a prompt. Write them for the model: say what went wrong, what the valid inputs are, and what to try instead. This is cheaper than any amount of system-prompt engineering, and the effect is immediate.

Bounded output per tool call. Hard caps on bytes, plus tools designed to return summaries or slices rather than whole artifacts. A search tool that returns the top 20 matches with line numbers beats a read_file on a file the model then has to scroll through.

A trace you can replay. Log the exact messages array before each model call, the raw response, each tool invocation and its output, token usage and latency. Store it as JSON lines keyed by run id. With that file you can replay a run against a new model version, build an eval set from real failures, and answer "why" without guessing.

Here is what the loop looks like once those four are in. It is still short, which is the point.

# agent_loop_v2.py — same loop, with the controls that survive production
import json, os, time, uuid
from pathlib import Path
import anthropic

MODEL = os.environ.get("ANTHROPIC_MODEL", "claude-sonnet-5-5")
client = anthropic.Anthropic()
TRACE_DIR = Path("traces"); TRACE_DIR.mkdir(exist_ok=True)

FINISH_TOOL = {
    "name": "finish",
    "description": "Call this exactly once when the task is complete. Put the final answer in `summary`.",
    "input_schema": {
        "type": "object",
        "properties": {"summary": {"type": "string"}, "confidence": {"type": "number"}},
        "required": ["summary"],
    },
}


class Budget:
    def __init__(self, steps=30, tokens=400_000, seconds=600):
        self.steps, self.tokens, self.seconds = steps, tokens, seconds
        self.used_steps = self.used_tokens = 0
        self.started = time.monotonic()

    def charge(self, usage):
        self.used_steps += 1
        self.used_tokens += usage.input_tokens + usage.output_tokens

    def exceeded(self):
        if self.used_steps >= self.steps: return "step limit"
        if self.used_tokens >= self.tokens: return "token budget"
        if time.monotonic() - self.started >= self.seconds: return "time limit"
        return None


def clip(text: str, limit: int = 12_000) -> str:
    if len(text) <= limit: return text
    head, tail = text[: limit // 2], text[-limit // 2 :]
    return f"{head}\n... [{len(text) - limit} characters omitted] ...\n{tail}"


def run_agent(task, tools, execute, system, budget=None):
    budget = budget or Budget()
    run_id = uuid.uuid4().hex[:8]
    trace = (TRACE_DIR / f"{run_id}.jsonl").open("a")
    log = lambda kind, **data: trace.write(json.dumps({"t": time.time(), "kind": kind, **data}) + "\n")

    messages = [{"role": "user", "content": task}]
    all_tools = [*tools, FINISH_TOOL]
    grace_turn = False  # True once the budget is gone: the model gets one last call to `finish`
    log("start", task=task, model=MODEL)

    while True:
        response = client.messages.create(
            model=MODEL, max_tokens=4096, system=system, tools=all_tools, messages=messages
        )
        budget.charge(response.usage)
        log("model", stop=response.stop_reason, usage=response.usage.model_dump(),
            content=[b.model_dump() for b in response.content])
        messages.append({"role": "assistant", "content": response.content})

        calls = [b for b in response.content if b.type == "tool_use"]
        for call in calls:
            if call.name == "finish":
                log("finish", **call.input)
                return {"status": "done", "run_id": run_id, **call.input}

        if grace_turn:
            log("stopped", reason="budget")
            return {"status": "stopped", "reason": "budget exhausted", "run_id": run_id}

        if not calls:
            # The model talked instead of finishing. Ask it to use the finish tool so we get structure.
            messages.append({"role": "user", "content": "If the task is complete, call `finish`. Otherwise continue."})
        else:
            results = []
            for call in calls:
                started = time.monotonic()
                output = clip(execute(call.name, call.input))
                log("tool", name=call.name, input=call.input, output=output,
                    ms=int((time.monotonic() - started) * 1000))
                results.append({"type": "tool_result", "tool_use_id": call.id, "content": output})
            messages.append({"role": "user", "content": results})

        reason = budget.exceeded()
        if reason:
            log("budget", reason=reason)
            messages.append({"role": "user", "content": f"Stopping: {reason}. Call `finish` with what you have."})
            grace_turn = True

The finish tool deserves a comment. Letting the model end the run by simply not calling a tool is convenient, but it gives you unstructured text and no signal about confidence. Making completion an explicit tool call turns the end of every run into data: a summary field you can display, a confidence number you can route on (below 0.6, send to a human), and a clean place to attach validation before you accept the result.

A few things I believed and no longer do#

"A bigger context window means I can stop worrying about tool output." The current models take a million tokens. You still cannot afford to fill them on every step, because you pay for the whole window on every call and quality degrades long before the hard limit. The loop makes the cost compounding: step 30 re-reads everything from steps 1 to 29.

"Parallel tool calls are an optimisation." They are also a correctness trap. If the model asks to write_file and run_tests in the same turn, you have to decide the order and whether the second should see the first. I execute calls sequentially in the order the model emitted them unless a tool is marked read-only.

"The model should never see raw errors." The opposite. Sanitise secrets, yes. But the full stderr of a failing build is exactly the information the model needs, and a human-friendly "something went wrong" is useless to it.

"The loop is the easy part; the prompt is the hard part." Reverse that. A mediocre system prompt with good termination, good tool errors and a trace will beat a beautiful prompt in a loop that runs forever and cannot be debugged.

Where to go from here#

Everything that makes an agent reliable sits around this loop rather than inside the model: the tools and their permissions, the files that tell the model about your project, the way context is trimmed between steps, the checkpoints that let a run survive a crash, and the evals that tell you whether a change helped. The industry has settled on a word for that whole layer: the harness. That is the next post, and it is where the real engineering lives.

If you take one thing from this one: write the 40-line version yourself, once, with your own tools, before adopting a framework. You will understand every knob the framework exposes, and you will know which of its defaults to distrust.

References#

• Anthropic, Building effective agents (December 2024)
• Yao et al., ReAct: Synergizing Reasoning and Acting in Language Models (2022)
• Anthropic, Tool use with the Messages API