def agent(messages, tools):
    while True:
        resp = client.messages.create(
            model="claude-sonnet-4-6",
            max_tokens=2048,
            system="...",
            messages=messages,
            tools=tools,
        )
        messages.append({"role": "assistant", "content": resp.content})

        if resp.stop_reason == "end_turn":
            return resp                              # we're done

        results = []
        for block in resp.content:
            if block.type == "tool_use":
                out = dispatch(block.name, block.input)
                results.append({
                    "type": "tool_result",
                    "tool_use_id": block.id,
                    "content": out,
                })
        messages.append({"role": "user", "content": results})

~20 lines. Everything else in this module is a layer that sits above this loop or wires more tools into it.

"Claude Code is a tool-use loop with a thick harness." The same recipe applies to building your own agentic app — you don't need to invent it from scratch.

# Inside the parent agent's loop, the model decides to call the Task tool:
{
  "type": "tool_use",
  "name": "Task",
  "input": {
    "description": "Find all HTTP route definitions",
    "subagent_type": "explore",
    "prompt": "Search the codebase for Flask/FastAPI/Express route decorators.
               Return a list of (file, line, route) tuples. Don't summarise the
               implementations — just the routes."
  }
}
# The harness runs a brand-new Claude session with that prompt + the
# Explore agent's tools. Parent receives only the final report.

A serious agent uses several at once. Claude Code reads CLAUDE.md (project), ~/.claude/memory/*.md (user), and the running session (conversation) on every turn — three scopes, one composite prompt.

def route(task: str) -> str:
    # 1. Cheap classifier — Haiku, no thinking, low effort
    kind = call(
        model="claude-haiku-4-5",
        system="Classify the task as trivial / reasoning / general.",
        user=task,
        schema=TaskKind,
        thinking=False, effort="low",
    )

    # 2. Route the actual work to the right model
    if kind == "trivial":
        return call(model="claude-haiku-4-5", user=task)
    if kind == "reasoning":
        return call(model="claude-opus-4-7", user=task, effort="high")
    return call(model="claude-sonnet-4-6", user=task)

"Route quality where it matters; route speed where it doesn't." Also works with fallbacks (primary fails → try secondary) and ensembles (run two, pick best).

# Agent kicks off work it can't wait on
job_id = submit_task(
    prompt="Analyse this 2GB log file",
    callback_url="https://my.app/agent/resume",
)
return f"Started job {job_id}, will resume when done."

# Later, external worker finishes and POSTs back:
@app.route("/agent/resume", methods=["POST"])
def resume():
    result = request.json["result"]
    continue_agent_with(result)

Stateful agents need a way to pause and resume. The architecture choice is: where does the state live while the agent is paused?

@app.route("/agent")
def agent():
    def stream():
        for ev in run_agent(request.args["q"], TOOLS):
            yield f"data: {json.dumps(asdict(ev))}\n\n"
    return Response(stream(),
                    mimetype="text/event-stream")

from dataclasses import dataclass, asdict
from typing import Iterator

@dataclass
class Progress:
    step: str          # shown to the user
    done: bool = False

def run_agent(user_msg, tools) -> Iterator[Progress]:
    messages = [{"role": "user", "content": user_msg}]
    while True:
        resp = client.messages.create(
            model="claude-sonnet-4-6", max_tokens=1024,
            tools=tools, messages=messages,
        )
        messages.append({"role": "assistant",
                         "content": resp.content})

        calls = [b for b in resp.content
                 if b.type == "tool_use"]
        if not calls:                  # model is done
            yield Progress("Done", done=True)
            return

        results = []
        for call in calls:
            yield Progress(f"Calling {call.name}…")  # ← before the slow part
            output = dispatch(call.name, call.input)
            results.append({"type": "tool_result",
                            "tool_use_id": call.id,
                            "content": output})
        messages.append({"role": "user", "content": results})

Emitting before each tool call is what lets the user watch the agent's reasoning unfold. The same generator works for any long operation — an indexing job, a batch run, a shell process — and never knows whether it's feeding a CLI, a web UI, or a log file. That's the point.

Agents fail interestingly. The risk isn't usually one disastrous tool call — it's 10,000 small ones that quietly burn budget, fill logs, or thrash on a bad assumption. Hard limits catch the latter.

You can build a capable agent with everything above. But shipping it to real users takes more than architecture — you have to know it works and keep it safe. That's next.

Subagents, memory, routing, and caching are all layers wrapped around this loop.

you › Top product by revenue last quarter?
  → list_tables() → describe_table(order_items) → run_query(SELECT …)
  Robot Vacuum — $46,307 (most recent quarter, completed orders).

you › What about the quarter before that?  # ← follow-up uses the history
  → run_query(SELECT …)
  Robot Vacuum again — $29,637, so revenue jumped ~56%.

Spot the two safety patterns from this module already in place: read-only by construction (the query tool opens the DB in mode=ro, so SQLite rejects any write) and a hard step budget (MAX_STEPS) so a confused agent can't loop forever.

"The agent loop is ~20 lines; everything else is harness." Managed Agents is Anthropic shipping the harness. For a self-contained CLI, the loop you own is the right call — reach for Managed Agents when you want hosted execution, versioned configs, file mounts, or server-managed sessions.

from tools import dispatch          # the SAME read-only tools.py, unchanged

# ── once: persist a versioned agent (model/system/tools live here, not on the session) ──
agent = client.beta.agents.create(
    name="Data Analyst", model="claude-sonnet-4-6",
    system="Inspect the schema before querying…",
    tools=CUSTOM_TOOLS,             # list_tables / describe_table / run_query, each type:"custom"
)

# ── per run: no while-loop — react to the event stream ──
session = client.beta.sessions.create(agent=agent.id, environment_id=env.id)
with client.beta.sessions.events.stream(session_id=session.id) as stream:
    client.beta.sessions.events.send(session_id=session.id, events=[
        {"type": "user.message", "content": [{"type": "text", "text": question}]}])
    for event in stream:
        if event.type == "agent.custom_tool_use":
            out = dispatch(event.name, event.input)     # ← our tools, our machine, our mode=ro gate
            client.beta.sessions.events.send(session_id=session.id, events=[
                {"type": "user.custom_tool_result",
                 "custom_tool_use_id": event.id,
                 "content": [{"type": "text", "text": out}]}])
        elif event.type == "session.status_idle" and event.stop_reason.type != "requires_action":
            break

Multi-turn is now free — the session is stateful server-side, so the messages list we hand-maintained disappears. (Alternatively: mount data.db as a file and let the built-in toolset query it inside Anthropic's sandbox — where the read-only file mount replaces our mode=ro gate.)