Lab 04 — A tool-heavy research assistant with the Claude Agent SDK

Lesson 04 · Building Your Own Agents · ~40 min hands-on

Goal: at the end of this lab you will have (a) run agent.py against a research question and seen a multi-turn trace, (b) inspected the structured LAST_REPORT output, (c) triggered parallel tool calls deliberately, and (d) modified one tool to see how the agent reacts to a new failure mode. This is the Claude Agent SDK’s “sweet spot” shape — a single agent with many tools doing a tool-heavy task.

Prereqs

• The agents/requirements.txt already installed (which pins anthropic>=0.39,<0.40). If not, run pip install -r agents/requirements.txt from the repo root.
• ANTHROPIC_API_KEY exported in your shell.
• L03 lab completed — you should be comfortable with Agent, @tool, and max_cost_usd.
• Working directory at the repo root (the paths below assume that).

Step 1 — Read agents/lesson04-tool-heavy-agent/agent.py

The full program lives in the repo at agents/lesson04-tool-heavy-agent/agent.py. Below it is reproduced inline so you can read along without flipping windows. We’ll then walk through each block.

Open the file in your editor — you don’t need to retype anything; the file is already there. The inlined copy in this lab guide is for reading convenience.

Step 2 — The full program

Here is the entire agent.py, verbatim:

"""Tool-heavy research assistant — Lesson 04 of the Agents course.

Built around the Claude Agent SDK primitives shipped inside the
`anthropic` Python package (>=0.39,<0.40).

Five mocked tools:
  - pkb_search / pkb_fetch  : in-process personal knowledge base
  - web_search / web_fetch  : in-process mock web search
  - write_report            : structured final-answer tool

Run:
    export ANTHROPIC_API_KEY=...
    python agent.py "What changed in Linux kernel 6.14's networking stack?"

NOTE: this file is also inlined in lab_guide.html. If you edit it,
also update the inlined block there.
"""
from __future__ import annotations

import sys
from typing import Any

from anthropic import Anthropic
from anthropic.agents import Agent, tool

# --- Mocked data --------------------------------------------------------

PKB: dict[str, dict[str, str]] = {
    "kernel-6.14-net-notes": {
        "title": "Personal notes on Linux 6.14 networking",
        "snippet": "io_uring zero-copy TX, BIG TCP for IPv4 by default.",
        "text": (
            "Linux kernel 6.14 ships io_uring zero-copy TX for TCP sockets, "
            "enables BIG TCP for IPv4 by default (was IPv6-only), and adds a "
            "new BPF kfunc family for tc-bpf flow tracking."
        ),
    },
    "kernel-6.14-misc": {
        "title": "Kernel 6.14 misc highlights",
        "snippet": "Bcachefs stable, sched_ext improvements.",
        "text": "Bcachefs is marked stable. sched_ext gets richer hooks.",
    },
}

WEB_INDEX: dict[str, list[dict[str, str]]] = {
    "linux 6.14 networking": [
        {
            "title": "LWN: kernel 6.14 networking changes",
            "url": "https://lwn.net/Articles/6.14-net/",
            "snippet": (
                "Notable changes: io_uring zero-copy TX, BIG TCP for IPv4 "
                "by default, new tc-bpf kfuncs for flow tracking."
            ),
        },
        {
            "title": "Kernelnewbies: 6.14 release notes",
            "url": "https://kernelnewbies.org/Linux_6.14",
            "snippet": "Networking: BIG TCP IPv4 default, MPTCP path-manager v2.",
        },
    ],
}

# --- Tools --------------------------------------------------------------

@tool
def pkb_search(query: str) -> str:
    """Search the personal knowledge base by substring.

    Returns one line per hit: '{doc_id}: {title} - {snippet}'.
    Use this BEFORE web_search to check internal notes.
    """
    q = query.lower()
    hits = [
        f"{k}: {v['title']} - {v['snippet']}"
        for k, v in PKB.items()
        if q in v["text"].lower() or q in v["title"].lower()
    ]
    return "\n".join(hits) if hits else "no matches"


@tool
def pkb_fetch(doc_id: str) -> str:
    """Fetch the full text of a PKB doc by id (as returned by pkb_search)."""
    doc = PKB.get(doc_id)
    return doc["text"] if doc else f"unknown doc: {doc_id}"


@tool
def web_search(query: str) -> str:
    """Search the web. Returns up to 3 hits as 'title | url | snippet' lines."""
    hits = WEB_INDEX.get(query.lower(), [])
    if not hits:
        # Fall back to substring match across all indexed queries.
        for k, v in WEB_INDEX.items():
            if any(word in k for word in query.lower().split()):
                hits = v
                break
    if not hits:
        return "no web results"
    return "\n".join(f"{h['title']} | {h['url']} | {h['snippet']}" for h in hits)


@tool
def web_fetch(url: str) -> str:
    """Fetch the text of a URL (returns the snippet from the search index)."""
    for hits in WEB_INDEX.values():
        for h in hits:
            if h["url"] == url:
                return h["snippet"]
    return f"could not fetch {url}"


@tool
def write_report(intro: str, bullets: list[str], sources: list[str]) -> str:
    """Save the final structured report.

    Call this exactly once, when you have enough material. Then give a
    one-sentence confirmation on the next turn to end the conversation.

    Args:
        intro: 1-2 sentence framing of the answer.
        bullets: 3 short bullet findings.
        sources: doc_ids or URLs the findings rely on.
    """
    LAST_REPORT.clear()
    LAST_REPORT.update({"intro": intro, "bullets": bullets, "sources": sources})
    return "saved"


# Module-level scratchpad so callers can grab the structured result after
# `agent.run()` returns. In production you'd hand a callback into the
# `write_report` tool factory; this is the lesson-friendly shape.
LAST_REPORT: dict[str, Any] = {}

# --- Agent assembly -----------------------------------------------------

SYSTEM_PROMPT = (
    "You are a research assistant. To answer a question:\n"
    "1. Check `pkb_search` first; fetch any promising doc with `pkb_fetch`.\n"
    "2. If you need more, use `web_search`; fetch with `web_fetch`.\n"
    "3. When you have enough material, call `write_report(intro, bullets, sources)`.\n"
    "Always cite sources by URL or doc_id. Be concise."
)


def log_step(step: Any) -> None:
    tools = [c.name for c in step.tool_calls]
    print(
        f"[step {step.turn}] stop_reason={step.stop_reason} "
        f"tools={tools} cost=${step.cost_so_far_usd:.4f}"
    )


def run_research_agent(question: str) -> dict[str, Any]:
    """Run the research assistant and return the structured report."""
    agent = Agent(
        client=Anthropic(),
        model="claude-3-5-sonnet-latest",
        system=SYSTEM_PROMPT,
        tools=[pkb_search, pkb_fetch, web_search, web_fetch, write_report],
        max_turns=15,
        max_cost_usd=0.50,
        on_step=log_step,
    )
    result = agent.run(question)
    return {
        "final_text": result.final_text,
        "cost_usd": result.usage.total_cost_usd,
        "turns": result.turns,
        "report": dict(LAST_REPORT),
    }


# --- CLI ----------------------------------------------------------------

if __name__ == "__main__":
    if len(sys.argv) < 2:
        sys.exit("usage: python agent.py 'your research question'")
    out = run_research_agent(sys.argv[1])
    print("\n=== final text ===")
    print(out["final_text"])
    print("\n=== structured report ===")
    for k, v in out["report"].items():
        print(f"{k}: {v}")
    print(f"\n[turns: {out['turns']}, cost: ${out['cost_usd']:.4f}]")

Step 3 — Run it

From the repo root:

python agents/lesson04-tool-heavy-agent/agent.py "What changed in Linux kernel 6.14's networking stack?"

Expected output: a 4–6-turn trace (one [step N] log line per turn), ending in a === final text === confirmation, followed by a === structured report === block showing the intro, bullets, and sources the LLM passed to write_report. The total cost printed at the end should be roughly $0.02–$0.05 with Sonnet.

✓ Self-check: you saw at least three [step N] lines, a final-text answer, and a populated structured report. No traceback.

Step 4 — Walk through the file

A guided tour, top to bottom:

• PKB and WEB_INDEX — the mocked data the agent searches. Adding a doc means adding a PKB[...] = {...} entry. The lesson keeps two PKB docs and one web query so the agent has a reason to escalate from the internal source to the external one.
• The five @tool functions — each is a normal Python function with a docstring. The docstring is what the LLM reads. Notice pkb_search’s docstring explicitly says “BEFORE web_search” — that nudges the agent toward the cheap source first.
• write_report — note that it doesn’t return the report; it stuffs it into module-level LAST_REPORT and returns "saved". This cleanly separates “the LLM’s view” (the tool succeeded) from “your view” (the structured payload).
• SYSTEM_PROMPT — three numbered instructions. Notice it doesn’t prescribe an exact tool order — only the rough policy. The LLM decides which tools to call in which order given the question.
• log_step — the per-turn observability hook. In production this would write to a structured logger or an event sink (e.g. an OTEL span).
• run_research_agent — the entry point. Constructs the Agent, calls .run(), and returns the structured result. Caps are max_turns=15 and max_cost_usd=0.50.
• __main__ block — CLI wrapper, prints everything.

Step 5 — Trigger parallel tool calls

The Claude Agent SDK supports parallel tool calls when the model decides multiple calls can be issued together. To see it happen, edit the system prompt in run_research_agent to add this line at the end:

"If you have multiple URLs to fetch, request them in parallel."

Then re-run with a question that yields at least two URLs (the kernel 6.14 question hits two web results):

python agents/lesson04-tool-heavy-agent/agent.py "What changed in Linux kernel 6.14's networking stack?"

Expected: one of the [step N] log lines shows tools=['web_fetch', 'web_fetch'] — two tool calls issued in one turn.

Restore the original system prompt before moving on.

Step 6 — Trigger a tool error

Edit pkb_fetch to fail unconditionally:

@tool
def pkb_fetch(doc_id: str) -> str:
    """Fetch the full text of a PKB doc by id (as returned by pkb_search)."""
    raise RuntimeError("PKB temporarily unavailable")

Re-run the same research question. Expected: the agent sees the is_error: true tool result on the next turn and either (a) tries web_search instead, or (b) calls write_report with what it already had from pkb_search and admits the gap.

✓ Self-check: the agent doesn’t crash; it routes around the failure and still produces a structured report.

Restore the original pkb_fetch body before moving on.

Step 7 — Force the cost cap

Edit max_cost_usd=0.50 to max_cost_usd=0.001 in run_research_agent. Re-run.

Expected: the SDK raises MaxCostExceeded after one or two turns — the prompt and the first tool round-trip already exceed a penny’s worth of tokens. Note the typed exception name vs. the generic RuntimeError you’d hand-write in the L02 loop.

Restore max_cost_usd=0.50 before moving on.

Step 8 — Add a sixth tool

Add a quote_source tool that returns a formatted citation snippet. Drop this alongside the other @tool functions:

@tool
def quote_source(text: str) -> str:
    """Format `text` as a quotable citation snippet for the final report.

    Use this when you find a sentence in a source worth quoting verbatim.
    """
    return f'> "{text}"'

Add it to the agent’s tools=[...] list:

tools=[pkb_search, pkb_fetch, web_search, web_fetch, write_report, quote_source],

Update the system prompt to mention it — e.g., append:

"When a source has a sentence worth quoting verbatim, call `quote_source` first and use its return value in `bullets`."

Re-run. Expected: the LLM calls quote_source opportunistically when it has a quotable snippet, and the resulting bullets look more citation-ready.

What you should leave with

• agent.py running end-to-end against a research question.
• First-hand observation of parallel tool calls, tool errors, and cost caps.
• A structured report dict you can hand to any downstream code.
• The instinct that the Claude Agent SDK’s “you write the tools, it runs the loop” division of labour fits a tool-heavy problem like a glove.

Troubleshooting