A small agent that uses three tools

Everything in the previous lessons was one prompt, one tool call, one result. An agent is what happens when you remove the one-call constraint. The model calls tools repeatedly, in whatever order it decides, until it has enough information to answer. Your code does not decide when to stop. The model does. Your code's only job is to keep running until the model says it is done — or until you hit a safety limit you set yourself.

What an agent actually is

An agent is a while loop. It runs until the model returns stop_reason: "end_turn" with no tool_use blocks in the response. Each iteration: send the message history, receive the response, execute any tool calls, append results to the history, call the API again. The tools are the same handlers you built in earlier lessons. What changes is the driver that orchestrates them.

Two failure modes end an agent loop badly: the loop never terminates because the model keeps calling tools (unbounded loop), or the loop crashes because the dispatch table raises on an unknown tool name (uncaught KeyError). Both are preventable with two additions: a maximum iteration cap and a safe dispatch function.

The full agent

# scripts/research_agent.py
import anthropic, json, os, httpx

client = anthropic.Anthropic()
SANDBOX = os.path.realpath(os.path.join(os.path.dirname(__file__), "../workspace"))
os.makedirs(SANDBOX, exist_ok=True)

# == Tool definitions ==
tools = [
    {
        "name": "search_web",
        "description": (
            "Search the web and return a list of relevant results with titles and snippets. "
            "Use to find factual information, recent events, or source material. "
            "Returns up to 5 results."
        ),
        "input_schema": {
            "type": "object",
            "properties": {"query": {"type": "string"}},
            "required": ["query"]
        }
    },
    {
        "name": "read_file",
        "description": "Read a file from the workspace. Path is relative to workspace root.",
        "input_schema": {
            "type": "object",
            "properties": {"path": {"type": "string"}},
            "required": ["path"]
        }
    },
    {
        "name": "write_file",
        "description": "Write text to a file in the workspace. Creates or overwrites.",
        "input_schema": {
            "type": "object",
            "properties": {
                "path": {"type": "string"},
                "content": {"type": "string"}
            },
            "required": ["path", "content"]
        }
    }
]

# == Handlers ==
def handle_search_web(query: str) -> dict:
    # DuckDuckGo Instant Answer API — free, no key required, returns limited results
    try:
        r = httpx.get(
            "https://api.duckduckgo.com/",
            params={"q": query, "format": "json", "no_redirect": 1},
            timeout=8
        )
        r.raise_for_status()
        data = r.json()
        results = []
        for topic in (data.get("RelatedTopics") or [])[:5]:
            if "Text" in topic:
                results.append({"title": topic.get("Text", "")[:100],
                                "url": topic.get("FirstURL", "")})
        if not results and data.get("AbstractText"):
            results.append({"title": data["AbstractText"][:200],
                            "url": data.get("AbstractURL", "")})
        return {"results": results, "count": len(results)}
    except Exception as e:
        return {"error": str(e)}

def _safe_path(p: str) -> str:
    resolved = os.path.realpath(os.path.join(SANDBOX, p))
    if not resolved.startswith(SANDBOX + os.sep) and resolved != SANDBOX:
        raise ValueError(f"Path escapes sandbox: {p!r}")
    return resolved

def handle_read_file(path: str) -> dict:
    try:
        with open(_safe_path(path), "r", encoding="utf-8") as f:
            return {"content": f.read()}
    except FileNotFoundError:
        return {"error": f"Not found: {path}"}
    except ValueError as e:
        return {"error": str(e)}

def handle_write_file(path: str, content: str) -> dict:
    try:
        resolved = _safe_path(path)
        os.makedirs(os.path.dirname(resolved), exist_ok=True)
        with open(resolved, "w", encoding="utf-8") as f:
            f.write(content)
        return {"written": len(content.encode()), "path": path}
    except ValueError as e:
        return {"error": str(e)}

# == Dispatch table ==
HANDLERS = {
    "search_web": handle_search_web,
    "read_file": handle_read_file,
    "write_file": handle_write_file
}

def dispatch(name: str, input_dict: dict) -> dict:
    handler = HANDLERS.get(name)
    if handler is None:
        return {"error": f"Unknown tool: {name!r}"}
    try:
        return handler(**input_dict)
    except TypeError as e:
        return {"error": f"Bad arguments for {name}: {e}"}

# == Agent loop ==
MAX_ITERATIONS = 12

def run_agent(task: str) -> str:
    messages = [{"role": "user", "content": task}]

    for iteration in range(MAX_ITERATIONS):
        r = client.messages.create(
            model="claude-opus-4-5",
            max_tokens=1024,
            tools=tools,
            messages=messages
        )
        messages.append({"role": "assistant", "content": r.content})

        if r.stop_reason == "end_turn":
            text_blocks = [b for b in r.content if hasattr(b, "text")]
            return text_blocks[0].text if text_blocks else "(no text response)"

        if r.stop_reason != "tool_use":
            return f"Unexpected stop_reason: {r.stop_reason}"

        tool_results = []
        for block in r.content:
            if block.type == "tool_use":
                print(f"[iter {iteration+1}] {block.name}({json.dumps(block.input)[:120]})")
                result = dispatch(block.name, block.input)
                print(f"           → {json.dumps(result)[:120]}")
                tool_results.append({
                    "type": "tool_result",
                    "tool_use_id": block.id,
                    "content": json.dumps(result)
                })
        messages.append({"role": "user", "content": tool_results})

    return f"Agent hit iteration cap ({MAX_ITERATIONS}). Last response was tool_use."

if __name__ == "__main__":
    task = (
        "Search for three facts about the Aurora Borealis. "
        "Write a 150-word summary of what you find to workspace/aurora_summary.txt. "
        "Then read the file back and confirm the byte count."
    )
    print(run_agent(task))

What the loop does not do

The loop does not decide which tool to call, does not retry on its own, and does not stop early because the task looks finished. Those are the model's decisions. The loop's job is to execute what the model requests and keep the message history intact. The only thing the loop decides unilaterally is when to give up — that is the iteration cap.

The dispatch table pattern

HANDLERS maps string names to callables. dispatch() wraps the lookup with two error paths: unknown tool name (.get() returns None), and bad arguments (TypeError when **input_dict does not match the function signature). Both return structured errors, not raised exceptions. The model reads the error and can adapt — passing a corrected argument on the next call.

When you add a new tool, you add one entry to HANDLERS. The loop does not change. The dispatch function does not change. This is the correct separation: tools are data, the loop is infrastructure.

If you get stuck

Agent hits the iteration cap every run: The model is stuck in a pattern — often it keeps searching because the search results are thin. Add a constraint to the system prompt or task description: "If search results are insufficient after two attempts, write a summary based on what you have found and stop." One sentence breaks the loop reliably.

TypeError in dispatch because the model passed an unexpected keyword: The model extrapolated a parameter that is not in the schema. Add "additionalProperties": false to the input_schema object. This causes a schema validation error before the call reaches your handler, and the model will correct the argument on the next attempt.

Tool results are too long and the context window fills up: Truncate long results in the handler. search_web already caps at 5 results with 100-character snippets. For read_file, add a max byte count (content[:8000]) and append a truncation notice to the returned string.

Model calls write_file before searching: The task description did not make the dependency explicit. Rewrite the task: "Search first. Write only after you have found at least two facts." The model follows explicit sequencing instructions reliably.

Cross-reference

The safe_path and write handler here are simplified from l6 — the audit logging is stripped. The full instrumented version of this agent, with CallRecord capture and a pytest eval suite, is in l8.

Deliverable

scripts/research_agent.py — the complete script above, runnable with python scripts/research_agent.py. The agent must complete the Aurora Borealis task: search, write, read back, confirm byte count. The printed output must show at least two [iter N] lines (one search call and one write call at minimum) and end with a non-empty summary paragraph.