The loop: an agent in one file
The concepts chapter defined an agent as a model calling tools in a loop, with the model choosing each next step. This chapter turns that sentence into a program you can run: no framework, no cloud, one file of standard-library Python. By the end you will have watched the checkout-latency investigation from the concepts chapter execute on your own machine, and you will own every line that made it happen.
One design decision up front, because it shapes the whole file. The loop talks to the model through a small interface, and this lab plugs a scripted stand-in into that interface instead of a real model API. The stand-in plays back a fixed sequence of turns, like a tape recorder; it has no judgment at all. That is not a limitation to apologize for, it is the point: the loop cannot tell the difference, which proves the loop is independent of the intelligence plugged into it. It also makes the lab free, offline, and perfectly reproducible, and it is the same trick (swapping the model for a recording) that becomes replay debugging in Chapter 10. At the end of this chapter, the same loop runs against real Claude, and the diff is a constructor call.
The conversation is a list
Everything the model ever sees arrives as a list of messages, so that
list is our first data structure. Each message is a dictionary with a
role saying who is speaking:
user: the task, and later, tool results being reported back,assistant: what the model said, including any tool requests,tool: our convenience wrapper for delivering tool results (real APIs fold these into ausermessage; the shape differs, the idea does not).
The loop's whole job is to grow this list: append what the model said, append what the tools returned, hand the longer list back. Recall from the concepts chapter that the model is stateless; this list is the agent's short-term memory, rebuilt and re-sent on every call.
The model interface
Three tiny classes define the boundary between "our code" and "the thing that thinks":
@dataclass
class ToolCall:
id: str # unique per call, so results can be matched to requests
name: str
args: dict
@dataclass
class ModelTurn:
stop_reason: str # "tool_use" or "end_turn"
text: str = ""
tool_calls: list[ToolCall] = field(default_factory=list)
A ModelTurn is one model response: some text, maybe some tool requests,
and a stop reason, the machine-readable field that tells the loop what
the model expects to happen next. Real APIs have more stop reasons (out
of output room, refused, hit a stop sequence); our stand-in needs only
the two that drive the core loop, and the loop treats the stop reason as
the only signal it trusts. Text is for humans; the stop reason is the
contract.
Notice ToolCall.id. When a model requests two tools in one turn (ours
will), the results must be delivered back labeled, so the model knows
which result answers which request. The id is that label. Losing track of
it is one of the classic hand-rolled-loop bugs.
The stand-in that implements the interface is eleven lines:
class ScriptedModel:
def __init__(self, turns: list[ModelTurn]):
self._turns = list(turns)
def respond(self, messages: list[dict]) -> ModelTurn:
if not self._turns:
raise RuntimeError("script exhausted: ...")
return self._turns.pop(0)
It ignores messages entirely, which is exactly what makes it a tape
recorder rather than a mind. The script it plays, three turns long, is
the investigation you already know: look at metrics, then check deploys
and logs in parallel, then conclude.
The world
An investigation needs something to investigate, so the file carries a
tiny canned world: hourly p99 latency for a checkout service that
jumps 14x at 02:00, three log lines about a payments-db connection
pool, and two deployments, one of which is a config change at 01:55 that
shrank that pool from 50 connections to 10. Three read-only tools expose
it: query_metrics, search_logs, list_deploys. Each tool is a plain
Python function plus a Tool record holding its name, description, and
parameter schema; the next chapter is entirely
about why those descriptions matter, so here we keep them short and move
on.
The loop itself
Here is the heart of the file, the part the whole book grows from:
def run_agent(model, tools: list[Tool], task: str, max_turns: int = 10) -> str:
toolbox = {t.name: t for t in tools}
messages: list[dict] = [{"role": "user", "content": task}]
for turn in range(1, max_turns + 1):
reply = model.respond(messages) # 1. think
messages.append({"role": "assistant", ...}) # 2. remember
if reply.stop_reason == "end_turn": # 3. done?
return reply.text
results = []
for call in reply.tool_calls: # 4. act
tool = toolbox.get(call.name)
output = tool.fn(**call.args) if tool else "error: ..."
results.append({"tool_call_id": call.id, "content": output})
messages.append({"role": "tool", "results": results}) # 5. report
raise RuntimeError(f"gave up after {max_turns} turns")
Read it as the state machine from the concepts chapter. Each iteration: think (one stateless model call carrying the whole conversation), remember (append the reply, or it never happened), check the stop reason (the only exit that means success), act (execute every requested tool), report (append all results, labeled by id, in one message). Then around again.
Three details are load-bearing:
max_turnsis a real safety device, not politeness. A model stuck in a tool-calling rut will loop forever, and every iteration costs tokens. This is the first, crudest budget in the book; Part 6 replaces it with a real ledger, but no loop should ever ship without at least this.- Tool failures become messages, not exceptions. An unknown tool or
bad arguments produce an
error: ...string that goes back to the model as a result. The model reads it and adapts. Crashing the loop because the model asked for a misspelled tool would hand control of your process's lifetime to the least reliable component in it. - All results for a turn travel together. The model asked for deploys and logs in one turn; it gets both answers in one message. Splitting them across messages teaches a real model, over time, that parallel requests come back mangled, and it will stop making them.
Don't be confused: "the model called a tool." Everyone says it, and this book will too, but keep the truth in view: the model emitted a request, and the loop called the tool. The distinction sounds pedantic until you need it: every security property in Part 8 lives in that gap, because the loop can inspect, veto, log, or gate any request before it becomes an action. A model cannot call anything. It can only ask.
Run it
python3 loop_agent.py
=== task: Checkout p99 latency spiked overnight. Find the likely cause.
[turn 1] model: Starting with the latency numbers for checkout.
[turn 1] tool : query_metrics(service='checkout', window='24h')
[turn 1] -> {"service": "checkout", "window": "24h", "p99_ms": {"22:00": 210, "23:00": 205, "00:00": 214, "01:00": 209, "02:00": 2930, "03:00": 2870, "04:00": 2910}}
[turn 2] model: p99 jumps from ~210ms to ~2900ms at 02:00. Checking what changed and what the logs say around that time.
[turn 2] tool : list_deploys(service='checkout')
[turn 2] -> {"service": "checkout", "deploys": [{"service": "checkout", "version": "v840", "at": "18:20", "note": "copy change on receipt email"}, {"service": "checkout", "version": "v841", "at": "01:55", "note": "config: payments-db pool size 50 -> 10"}]}
[turn 2] tool : search_logs(query='payments-db')
[turn 2] -> {"query": "payments-db", "hits": ["02:03:11 ERROR TimeoutError: waiting for payments-db connection", "02:03:14 ERROR pool exhausted: 10/10 connections in use (payments-db)", "02:04:02 ERROR TimeoutError: waiting for payments-db connection"]}
[turn 3] model: Likely cause: deploy v841 at 01:55 shrank the payments-db connection pool from 50 to 10 (the note says it was a config change). Minutes later the logs show 'pool exhausted: 10/10' and connection timeouts, and p99 jumps 14x at 02:00. Recommend rolling back v841 or restoring pool size 50.
=== done: 3 turns
Read the transcript the way you will later read production traces. Turn 1 is a single tool call. Turn 2 is a parallel turn: two requests, two labeled results, one reporting message. Turn 3 cites its evidence: the deploy note, the log line, the metric jump, all three of which really appeared earlier in the transcript. That evidence chain is what verifiable agent output looks like, and Part 6's verifier agents will check chains like it mechanically.
The loop function is 33 lines. With the model interface and the printing it rounds to about a hundred, which was the promise, and there is nothing else: no framework, no hidden runtime. Every agent in this book, up to the five-hundred-worker fleets, is this loop wearing better infrastructure.
The same loop against real Claude
Swapping the tape recorder for a mind is a follow-along (it needs the
anthropic package and an API key, so its output below is illustrative
rather than captured). The structure maps one to one; the real API's
message shapes are slightly richer, and the tool results ride in a
user message:
import anthropic
client = anthropic.Anthropic() # reads ANTHROPIC_API_KEY from the env
api_tools = [{"name": t.name, "description": t.description,
"input_schema": t.params} for t in TOOLS]
toolbox = {t.name: t for t in TOOLS}
messages = [{"role": "user", "content":
"Checkout p99 latency spiked overnight. Find the likely cause."}]
while True:
response = client.messages.create(
model="claude-opus-4-8",
max_tokens=16000,
tools=api_tools,
messages=messages,
)
messages.append({"role": "assistant", "content": response.content})
if response.stop_reason != "tool_use":
break
results = [{"type": "tool_result", "tool_use_id": block.id,
"content": toolbox[block.name].fn(**block.input)}
for block in response.content if block.type == "tool_use"]
messages.append({"role": "user", "content": results})
(illustrative) The real model typically runs the same three-beat
investigation: metrics first, then deploys and logs, then a conclusion
naming v841, in 3 to 5 turns depending on how it groups its calls.
On AWS, the same code speaks to Bedrock by changing the client and the
model id (Claude models on Bedrock carry the anthropic. prefix; Part 2
covers regions, profiles, and everything else this snippet is silent
about):
from anthropic import AnthropicBedrockMantle
client = AnthropicBedrockMantle(aws_region="us-east-1")
# then exactly as above, with model="anthropic.claude-opus-4-8"
That is the payoff of the interface decision: scripted, first-party, or Bedrock, the loop does not change. Which means everything we build onto the loop from here (recording, replay, budgets, fleets) works against all three.
Full source
"""Lab 1.1: the agent loop, with no framework and no cloud.
One file, three parts:
1. a tiny world for the agent to investigate (canned metrics, logs,
deploys for a checkout service that got slow overnight),
2. a Model interface with a ScriptedModel stand-in, so the lab runs
deterministically with no API key,
3. the loop itself: call the model, execute the tools it asks for,
append results, repeat until it stops asking.
Standard library only. Deterministic: same run, same transcript.
"""
from __future__ import annotations
import json
from dataclasses import dataclass, field
from typing import Callable
# ---------------------------------------------------------------------------
# 1. The world: three read-only tools over canned data
# ---------------------------------------------------------------------------
METRICS = {
"checkout": {
"22:00": 210, "23:00": 205, "00:00": 214, "01:00": 209,
"02:00": 2930, "03:00": 2870, "04:00": 2910,
},
}
LOG_LINES = [
"02:03:11 ERROR TimeoutError: waiting for payments-db connection",
"02:03:14 ERROR pool exhausted: 10/10 connections in use (payments-db)",
"02:04:02 ERROR TimeoutError: waiting for payments-db connection",
]
DEPLOYS = [
{"service": "checkout", "version": "v840", "at": "18:20",
"note": "copy change on receipt email"},
{"service": "checkout", "version": "v841", "at": "01:55",
"note": "config: payments-db pool size 50 -> 10"},
]
def query_metrics(service: str, window: str) -> str:
"""Return p99 latency (ms) by hour for a service."""
series = METRICS.get(service)
if series is None:
return f"error: unknown service '{service}'"
return json.dumps({"service": service, "window": window, "p99_ms": series})
def search_logs(query: str) -> str:
"""Full-text search over recent application log lines."""
hits = [line for line in LOG_LINES if query.lower() in line.lower()]
return json.dumps({"query": query, "hits": hits})
def list_deploys(service: str) -> str:
"""List recent deployments for a service, newest last."""
rows = [d for d in DEPLOYS if d["service"] == service]
return json.dumps({"service": service, "deploys": rows})
@dataclass
class Tool:
name: str
description: str
params: dict # JSON-Schema-shaped parameter spec
fn: Callable[..., str]
TOOLS = [
Tool(
name="query_metrics",
description=("Read p99 latency by hour for one service. Use when "
"the question is how slow or how many."),
params={"type": "object",
"properties": {"service": {"type": "string"},
"window": {"type": "string"}},
"required": ["service", "window"]},
fn=query_metrics,
),
Tool(
name="search_logs",
description=("Full-text search over application log lines. Use "
"when hunting for error messages or exceptions."),
params={"type": "object",
"properties": {"query": {"type": "string"}},
"required": ["query"]},
fn=search_logs,
),
Tool(
name="list_deploys",
description=("List recent deployments with times and change notes. "
"Use when asking what changed recently."),
params={"type": "object",
"properties": {"service": {"type": "string"}},
"required": ["service"]},
fn=list_deploys,
),
]
# ---------------------------------------------------------------------------
# 2. The model interface, and a scripted stand-in
# ---------------------------------------------------------------------------
@dataclass
class ToolCall:
id: str
name: str
args: dict
@dataclass
class ModelTurn:
stop_reason: str # "tool_use" or "end_turn"
text: str = ""
tool_calls: list[ToolCall] = field(default_factory=list)
class ScriptedModel:
"""Plays back a fixed list of turns.
A stand-in for the real model API with the same interface, so the loop
code cannot tell the difference. Part 2 swaps in the real thing.
"""
def __init__(self, turns: list[ModelTurn]):
self._turns = list(turns)
def respond(self, messages: list[dict]) -> ModelTurn:
if not self._turns:
raise RuntimeError("script exhausted: the loop asked for more "
"turns than the script contains")
return self._turns.pop(0)
INVESTIGATION_SCRIPT = [
ModelTurn(
stop_reason="tool_use",
text="Starting with the latency numbers for checkout.",
tool_calls=[ToolCall("t1", "query_metrics",
{"service": "checkout", "window": "24h"})],
),
ModelTurn(
stop_reason="tool_use",
text=("p99 jumps from ~210ms to ~2900ms at 02:00. Checking what "
"changed and what the logs say around that time."),
tool_calls=[
ToolCall("t2", "list_deploys", {"service": "checkout"}),
ToolCall("t3", "search_logs", {"query": "payments-db"}),
],
),
ModelTurn(
stop_reason="end_turn",
text=("Likely cause: deploy v841 at 01:55 shrank the payments-db "
"connection pool from 50 to 10 (the note says it was a config "
"change). Minutes later the logs show 'pool exhausted: 10/10' "
"and connection timeouts, and p99 jumps 14x at 02:00. "
"Recommend rolling back v841 or restoring pool size 50."),
),
]
# ---------------------------------------------------------------------------
# 3. The loop
# ---------------------------------------------------------------------------
def run_agent(model, tools: list[Tool], task: str, max_turns: int = 10) -> str:
toolbox = {t.name: t for t in tools}
messages: list[dict] = [{"role": "user", "content": task}]
print(f"=== task: {task}")
for turn in range(1, max_turns + 1):
reply = model.respond(messages)
if reply.text:
print(f"[turn {turn}] model: {reply.text}")
messages.append({"role": "assistant", "content": reply.text,
"tool_calls": [vars(c) for c in reply.tool_calls]})
if reply.stop_reason == "end_turn":
print(f"=== done: {turn} turns")
return reply.text
results = []
for call in reply.tool_calls:
tool = toolbox.get(call.name)
if tool is None:
output = f"error: no tool named '{call.name}'"
else:
try:
output = tool.fn(**call.args)
except TypeError as exc:
output = f"error: bad arguments: {exc}"
args = ", ".join(f"{k}={v!r}" for k, v in call.args.items())
print(f"[turn {turn}] tool : {call.name}({args})")
print(f"[turn {turn}] -> {output}")
results.append({"tool_call_id": call.id, "content": output})
messages.append({"role": "tool", "results": results})
raise RuntimeError(f"gave up after {max_turns} turns without end_turn")
if __name__ == "__main__":
run_agent(
model=ScriptedModel(INVESTIGATION_SCRIPT),
tools=TOOLS,
task="Checkout p99 latency spiked overnight. Find the likely cause.",
)
👉 Next: tool design. The loop treats tools as interchangeable; the model does not. The words in a tool's description quietly decide whether your agent is precise or erratic, and the next lab measures exactly that.