AI Beyond Fundamentals › Module 6 › Lesson

Module 6 · Tool Calling at Scale

Tool Calling at Scale

Module 5 gave your agent hands. This lesson asks what the hands cost. Every tool result stays in the conversation and is read again on every later turn, so a task with dozens of calls can read hundreds of thousands of tokens to do very little thinking. You will run one real task two ways, measure both, and learn when to let the model write the loop instead of running it.

The task is real: curate a twelve-piece exhibit from the Metropolitan Museum of Art's open collection, with downloaded images, a contact sheet and a gallery page. Same six tools, same task, same model, two ways of calling them. It builds on tool calling from Module 2 and the loop, limits and sandbox from Module 5, and it needs Docker.

Concepts

Every tool result is read again on every later turn

A model remembers nothing between calls. Each turn you send the whole conversation again: the instructions, the tool list, every earlier call and every earlier result. So a tool result is not paid for once. It is paid for on every turn after it arrives. A museum record is about 2.2 KB of JSON, roughly 550 tokens. Fetch 30 of them on turn 3 and all 30 are read again on turn 4, turn 5 and every turn after that, whether or not the model still needs them.

40k 1.3k 1 2.4k 2 15.5k 3 17.0k 4 29.2k 5 36.6k 6 39.1k 7 41.4k 8 41.7k 9 42.3k 10 1.2k 1 4.8k 2 5.9k 3 Loop: one call at a time 266,378 tokens read in total Code: one script 11,859 tokens read in total model turn model turn
Input tokens the model read on each turn, one run of each mode: same task, same six tools, same model. The loop run climbs because every result stays in the conversation. The code run never passes 6,000, because the records stayed inside the sandbox.

The tool list is not the problem here. Six small tools cost about 1,200 tokens in either mode. With dozens of tools or long descriptions it becomes a second cost on every turn, and Anthropic's engineering team describes agents connected to thousands of tools reading hundreds of thousands of tokens before they see the request. The cost in this lesson comes from the other source: results piling up.

Count tokens read, not calls

It is tempting to count round trips, and many explanations of this problem do. But models batch independent calls. In our three loop runs the model made 52, 58 and 57 tool calls in 10, 10 and 9 turns, often a dozen at once. Calls overstate turns. What grows the bill is how much has piled up by each turn, so the honest measure is the sum of usage.input_tokens across the run, which is what Lab 2 logs.

Calls that depend on each other cannot be batched: search, then look at the results, then choose, then download the choices. Those chains are where turns genuinely accumulate. One more caveat: these runs switch prompt caching off so you can see what the model reads. With caching on, a repeated prefix is billed at a lower rate and returns faster, which narrows the cost gap. It does not shrink what the model has to hold in its context.

Let the model write the loop

In code mode the tools become ordinary Python functions, and the model writes one script that calls them: search, loop over the ids, fetch each record, filter, download, build the page. A sandbox runs the script and only what the script prints goes back to the model. The records never enter its context.

Anthropic's engineering team describes this approach for MCP servers and reports a Drive-to-Salesforce example dropping from about 150,000 tokens to 2,000, a 98.7% saving. That is their own worked example, not a benchmark. Ours is a different task with its own numbers: a median of 267,531 input tokens in loop mode against 11,859 in code mode, about 23 times fewer.

Tokens are not the only difference. Exactness: the requirement "twelve different kinds of object" is one line of code, len(set(kinds)) == 12. In all three loop runs the model fell short (9, 10 and 11 distinct kinds) after fetching 30 to 36 records and judging them close enough. One run said so itself. All three code runs passed. Exposure: bulky text from outside, here museum descriptions, stays in the sandbox instead of sitting in front of the model, and the model sees only what the script chooses to print.

Loop: one call at a timeCode: one script
Input tokens, median267,53111,859
Input tokens, range266,378 to 299,8398,798 to 36,124
Model turns9 to 103 to 7
Tool calls or scripts run52 to 58 calls2 to 6 scripts
Met every requirement0 of 3 runs3 of 3 runs
Wall-clock time33 to 35 s18 to 55 s

Three runs per mode, one task, one model (Claude Haiku 5.5), prompt caching off. Treat the shape as the finding, not the exact numbers: yours will differ, and Lab 4 has you measure your own.

A sandbox with the network on

Module 5's sandboxed tool ran with no network at all, so even a hostile script could not send anything out. These tools have to call a public API, so this container needs the network, and you cannot have both. The protection changes shape. Instead of "it cannot reach out", the rule becomes "there is nothing in it worth sending":

  • No secrets inside. No API keys, no home folder. The only environment variable passed in is the data folder's path.
  • One writable folder. The tools are mounted read-only; results land in a single data folder.
  • Limits enforced from outside. The timeout kills the whole container from the host, so the script cannot disable it. Memory, CPU and process counts are capped.
  • Least privilege. A non-root user, a read-only filesystem, and every Linux capability dropped.

What this does not stop: the script can still send whatever it can see to any server. That is acceptable only because it can see public data and nothing else. If your tools carry credentials or private data, use the stricter pattern: keep the container offline and send each call out through a small bridge on the host that allows only named tools. Anthropic's post lists sandboxing, resource limits and monitoring as the real cost of this approach, and they are.

When not to, and what goes wrong

  • One or two calls. A loop is simpler, and there is nothing to save.
  • Judgment between calls, or approval. If a person or a policy has to sign off before the next step, that belongs in the harness loop from Module 5, not inside a script that runs to the end.
  • Debugging gets harder. Our slowest code run took six scripts and 36,124 tokens because the first attempts failed. That was still far below any loop run, but four times the best code run. Code mode was not always faster on the clock either: 18 to 55 seconds against about 34 for the loop, once container start-up and retries count.
  • APIs change under you. While building this lesson, the Met retired its old search endpoint on 1 October 2026. The old URL answered HTTP 410 with a message naming the replacement. Because the tool wrapper hands that message back as text, the failure was readable and the fix took a minute. A tool that swallowed the error and returned an empty list would have produced an empty exhibit and no explanation. Lab 5 has you cause this on purpose.

Hands-on Labs

  1. Create a folder for this lesson inside your lab repo and work from inside it. This lesson adds two libraries; the model SDK and requests are already in your environment.
    mkdir toolscale && cd toolscale pip install pillow requests
  2. Create core.py. It knows nothing about museums, so a different example later reuses it unchanged. It gives you a disk cache (so every run sees identical data), a ToolError the model can read, a @tool decorator that builds each tool's schema from its type hints and docstring, and a call_tool that turns every failure into text.
    """Shared plumbing for the tool-calling-at-scale lab. Nothing in here knows about museums. A new example (Pokemon, recipes, ...) is a new tools file that imports from this one and decorates its functions with @tool. """ import hashlib import inspect import json import os import time import typing from pathlib import Path import requests ROOT = Path(__file__).parent # In the container the code is mounted read-only, so data lives somewhere writable. DATA_DIR = Path(os.environ.get("TOOLSCALE_DATA", ROOT)) CACHE_DIR = DATA_DIR / "cache" OUT_DIR = DATA_DIR / "out" USER_AGENT = "ai-beyond-fundamentals-lab/1.0 (course exercise)" MIN_GAP_SECONDS = 0.15 # be polite to free public APIs CACHE_DIR.mkdir(exist_ok=True) OUT_DIR.mkdir(exist_ok=True) _last_request = 0.0 stats = {"network_calls": 0, "cache_hits": 0} class ToolError(Exception): """A failure the model should be able to read and act on.""" # ---------------------------------------------------------------- HTTP + cache def _cache_path(key: str, suffix: str) -> Path: return CACHE_DIR / (hashlib.sha256(key.encode()).hexdigest()[:24] + suffix) def fetch(url: str, params: dict | None = None, *, binary: bool = False): """GET a URL with a disk cache, so every run (loop or code) sees identical data. Failures raise ToolError carrying the API's own message, never a bare status code. """ global _last_request key = url + "?" + json.dumps(params or {}, sort_keys=True) path = _cache_path(key, ".bin" if binary else ".json") if path.exists(): stats["cache_hits"] += 1 data = path.read_bytes() return data if binary else json.loads(data) wait = MIN_GAP_SECONDS - (time.time() - _last_request) if wait > 0: time.sleep(wait) try: resp = requests.get(url, params=params, timeout=20, headers={"User-Agent": USER_AGENT}) except requests.RequestException as exc: raise ToolError(f"network error calling {url}: {exc}") from exc finally: _last_request = time.time() stats["network_calls"] += 1 if resp.status_code != 200: detail = resp.text[:400] try: detail = resp.json().get("message", detail) except ValueError: pass raise ToolError(f"{url} returned HTTP {resp.status_code}: {detail}") path.write_bytes(resp.content) return resp.content if binary else resp.json() def safe_output_path(filename: str, ext: str) -> Path: """Model-chosen filenames are untrusted input: keep only the base name, force the extension.""" stem = Path(filename).stem or "output" stem = "".join(c for c in stem if c.isalnum() or c in "-_") or "output" return OUT_DIR / f"{stem}{ext}" # ------------------------------------------------------- tool registry + schema TOOLS: dict[str, typing.Callable] = {} _JSON_TYPES = {int: "integer", float: "number", str: "string", bool: "boolean"} def tool(fn: typing.Callable) -> typing.Callable: """Register a plain function as a tool. Its type hints and docstring ARE the contract.""" TOOLS[fn.__name__] = fn return fn def _json_type(hint) -> dict: origin = typing.get_origin(hint) if origin is list: (inner,) = typing.get_args(hint) return {"type": "array", "items": _json_type(inner)} if hint in _JSON_TYPES: return {"type": _JSON_TYPES[hint]} raise TypeError(f"unsupported tool argument type: {hint!r}") def schema_for(fn: typing.Callable) -> dict: """Build an Anthropic-format tool definition from a function's hints and docstring.""" sig = inspect.signature(fn) hints = typing.get_type_hints(fn) props, required = {}, [] for name, param in sig.parameters.items(): props[name] = _json_type(hints[name]) if param.default is inspect.Parameter.empty: required.append(name) return { "name": fn.__name__, "description": inspect.getdoc(fn) or "", "input_schema": {"type": "object", "properties": props, "required": required}, } def all_schemas() -> list[dict]: return [schema_for(fn) for fn in TOOLS.values()] def call_tool(name: str, args: dict) -> str: """Run a registered tool and return a string the model can read. Errors come back as text instead of raising, so the model sees them and can recover. """ if name not in TOOLS: return f"ERROR: unknown tool {name!r}. Available: {', '.join(TOOLS)}" try: result = TOOLS[name](**args) except ToolError as exc: return f"ERROR: {exc}" except TypeError as exc: return f"ERROR: bad arguments for {name}: {exc}" except Exception as exc: # a tool bug must not kill the loop; the model can read it and adapt return f"ERROR: {name} failed unexpectedly: {type(exc).__name__}: {exc}" return result if isinstance(result, str) else json.dumps(result, ensure_ascii=False)
  3. Three details are worth a slow read. fetch identifies itself with an honest User-Agent and waits briefly between requests; free public APIs are a shared resource. safe_output_path exists because a filename the model chooses is untrusted input: an earlier version of this code let a model-supplied name with no extension crash the whole run. And the last except in call_tool returns unexpected errors as text so the model can adapt instead of the loop dying.
  4. Create met_tools.py: six tools over the Metropolitan Museum of Art's open collection. The search endpoint is the paginated v1.1, and limit tops out at 500 per page. The docstrings are the contract the model reads, so write them for the model.
    """Museum exhibit curator tools, backed by the Met's open-access Collection API. Six tools, used identically by the loop runner and the code runner. """ import html import io from PIL import Image, ImageDraw from core import OUT_DIR, ToolError, fetch, safe_output_path, tool BASE = "https://collectionapi.metmuseum.org/public/collection" @tool def list_departments() -> list: """List the museum's departments as {departmentId, displayName}. Use a departmentId to narrow a search.""" return fetch(f"{BASE}/v1/departments")["departments"] @tool def search_collection(query: str, offset: int = 0, limit: int = 100, department_id: int = 0) -> dict: """Search the collection for public-domain objects that have images. Returns {total, object_ids, next_offset}. Results are paged: limit is at most 500, and next_offset is null once you have reached the end. department_id 0 means all. """ params = { "q": query, "hasImages": "true", "isPublicDomain": "true", "offset": offset, "limit": min(limit, 500), } if department_id: params["departmentId"] = department_id data = fetch(f"{BASE}/v1.1/search", params) ids = data.get("objectIDs") or [] nxt = offset + len(ids) return {"total": data.get("total", 0), "object_ids": ids, "next_offset": nxt if nxt < data.get("total", 0) and ids else None} @tool def get_object(object_id: int) -> dict: """Get the full record for one object: title, artist, date, medium, department, image URLs and more.""" return fetch(f"{BASE}/v1/objects/{object_id}") @tool def download_image(object_id: int) -> str: """Download the small web image for an object and return its local file path.""" obj = get_object(object_id) url = obj.get("primaryImageSmall") if not url: raise ToolError(f"object {object_id} has no image") path = OUT_DIR / f"{object_id}.jpg" path.write_bytes(fetch(url, binary=True)) return str(path) @tool def make_contact_sheet(object_ids: list[int], filename: str = "contact_sheet.png") -> str: """Tile the downloaded images for these objects into one labelled PNG and return its path. Each object must have been fetched with download_image first. """ cell, label_h, cols = 240, 22, 4 paths = [] for oid in object_ids: p = OUT_DIR / f"{oid}.jpg" if not p.exists(): raise ToolError(f"image for object {oid} not downloaded yet; call download_image first") paths.append((oid, p)) rows = -(-len(paths) // cols) sheet = Image.new("RGB", (cols * cell, rows * (cell + label_h)), "white") draw = ImageDraw.Draw(sheet) for i, (oid, p) in enumerate(paths): img = Image.open(p).convert("RGB") img.thumbnail((cell, cell)) x, y = (i % cols) * cell, (i // cols) * (cell + label_h) sheet.paste(img, (x + (cell - img.width) // 2, y)) draw.text((x + 4, y + cell + 4), f"#{oid}", fill="black") out = safe_output_path(filename, ".png") sheet.save(out) return str(out) @tool def build_gallery(title: str, object_ids: list[int], filename: str = "exhibit.html") -> str: """Write a self-contained HTML gallery page for these objects (image, title, artist, date) and return its path. Each object must have been fetched with download_image first. """ cards = [] for oid in object_ids: if not (OUT_DIR / f"{oid}.jpg").exists(): raise ToolError(f"image for object {oid} not downloaded yet; call download_image first") obj = get_object(oid) cards.append( f'<figure><img src="{oid}.jpg" alt="{html.escape(obj["title"])}">' f'<figcaption><strong>{html.escape(obj["title"])}</strong><br>' f'{html.escape(obj.get("artistDisplayName") or "Unknown artist")}, ' f'{html.escape(obj.get("objectDate") or "undated")}<br>' f'<a href="{html.escape(obj["objectURL"])}">View at the Met</a></figcaption></figure>' ) page = ( "<!DOCTYPE html><html lang='en'><meta charset='utf-8'>" f"<title>{html.escape(title)}</title>" "<style>body{font:16px system-ui;max-width:1100px;margin:2rem auto;padding:0 1rem}" ".g{display:grid;grid-template-columns:repeat(auto-fill,minmax(240px,1fr));gap:1.5rem}" "img{width:100%;height:240px;object-fit:contain;background:#f4f4f4}" "figcaption{font-size:14px;margin-top:.4rem}</style>" f"<h1>{html.escape(title)}</h1><p>Images and data: The Metropolitan Museum of Art, public domain.</p>" f"<div class='g'>{''.join(cards)}</div>" ) out = safe_output_path(filename, ".html") out.write_text(page, encoding="utf-8") return str(out)
  5. Create smoke.py: it checks the error paths, then does the whole exhibit as one plain script. Read this one closely, because in Lab 4 the model will write something like it.
    import json import core import met_tools as m # noqa: F401 (registers the tools) print("tools:", list(core.TOOLS)) print(json.dumps(core.schema_for(m.search_collection), indent=1)[:700], "\n") # Error paths come back as readable text print(core.call_tool("get_object", {"object_id": 99999999})) print(core.call_tool("get_object", {"nope": 1})) print(core.call_tool("not_a_tool", {}), "\n") # The whole workflow as ONE script: this is what code mode would write. ids, offset = [], 0 while offset is not None and len(ids) < 300: page = m.search_collection("cat", offset=offset, limit=100) ids += page["object_ids"] offset = page["next_offset"] print("search ->", len(ids), "ids of", page["total"]) objs = [m.get_object(i) for i in ids[:60]] chars = sum(len(json.dumps(o)) for o in objs) print(f"fetched {len(objs)} objects, {chars:,} chars, ~{chars // 4:,} tokens if the model read them all") picks = [o for o in objs if o["primaryImageSmall"] and o["isPublicDomain"]] picks.sort(key=lambda o: (o["department"], o["objectDate"])) picks = picks[:12] for o in picks: m.download_image(o["objectID"]) ids12 = [o["objectID"] for o in picks] print("sheet ->", m.make_contact_sheet(ids12)) print("page ->", m.build_gallery("Cats in the Met", ids12)) print("stats ->", core.stats)
  6. Run it.
    python smoke.py
Check: the error lines are readable sentences (for example ObjectID not found), the script prints about 30,000 estimated tokens for 60 records, and out/contact_sheet.png shows 12 labelled images. The first run takes about 30 seconds because it calls the network; run it again and it finishes in a second or two because everything comes from cache/.
  1. Create tasks.py: the one task both modes are given, and the one check both are judged by. The check reads the files the run produced, not the model's own summary.
    """The one task both runners are given, and the one check both are judged by.""" import re import shutil from core import OUT_DIR, fetch EXHIBIT_TASK = """Curate a 12-piece mini exhibit called "Cats in the Met". Requirements: - Every piece must be public domain and have an image. - Pieces must come from at least 4 different museum departments. - No two pieces may be the same kind of object (for example, not 12 bronze statuettes). - Download the image for each of the 12 pieces. - Make a contact sheet of the 12 pieces (filename "contact_sheet") and build the gallery page (filename "exhibit"), titled "Cats in the Met". When you are done, reply with one line per piece: object id, title, department.""" def reset_outputs() -> None: """Clear generated files so each run starts from the same empty state (the data cache stays).""" shutil.rmtree(OUT_DIR, ignore_errors=True) OUT_DIR.mkdir() def verify() -> dict: """Check the exhibit against the requirements, by reading the files the run produced.""" page = OUT_DIR / "exhibit.html" sheet = OUT_DIR / "contact_sheet.png" result = {"gallery_exists": page.exists(), "sheet_exists": sheet.exists()} if not page.exists(): return {**result, "ok": False} ids = [int(i) for i in re.findall(r'src="(\d+)\.jpg"', page.read_text())] base = "https://collectionapi.metmuseum.org/public/collection/v1/objects" objs = [fetch(f"{base}/{i}") for i in ids] result.update( pieces=len(ids), distinct_pieces=len(set(ids)), all_public_domain=all(o["isPublicDomain"] for o in objs), all_have_images=all(o["primaryImageSmall"] for o in objs), departments=len({o["department"] for o in objs}), distinct_object_names=len({o["objectName"] for o in objs}), ) result["ok"] = ( result["pieces"] == 12 and result["distinct_pieces"] == 12 and result["all_public_domain"] and result["all_have_images"] and result["departments"] >= 4 and result["distinct_object_names"] == 12 and result["sheet_exists"] ) return result
  2. Create loop_mode.py. It is the Module 5 loop, with two deliberate choices: prompt caching is off so you can see what the model reads, and every turn's input tokens are recorded. Tool calls in the same response all run before the next turn, which is how one turn can hold a dozen calls.
    """Loop mode: the model calls one tool per step and re-reads everything every turn. python loop_mode.py [--model claude-haiku-5-5] [--max-turns 60] No prompt caching on purpose: we want to count what the model reads each turn. """ import argparse import json import time import anthropic import core import met_tools # noqa: F401 (registers the tools) from tasks import EXHIBIT_TASK, reset_outputs, verify SYSTEM = "You are a museum curator's assistant. Use the tools to do the task. Be efficient." def run(task: str, model: str, max_turns: int) -> dict: client = anthropic.Anthropic() tools = core.all_schemas() messages = [{"role": "user", "content": task}] turns, calls, total_in, total_out = [], 0, 0, 0 started = time.time() final_text, stop = "", "max_turns" for n in range(1, max_turns + 1): resp = client.messages.create( model=model, max_tokens=4096, system=SYSTEM, tools=tools, messages=messages, thinking={"type": "disabled"}, ) total_in += resp.usage.input_tokens total_out += resp.usage.output_tokens uses = [b for b in resp.content if b.type == "tool_use"] turns.append({"turn": n, "input_tokens": resp.usage.input_tokens, "output_tokens": resp.usage.output_tokens, "tools": [u.name for u in uses]}) messages.append({"role": "assistant", "content": resp.content}) if not uses: final_text = "".join(b.text for b in resp.content if b.type == "text") stop = resp.stop_reason break results = [] for u in uses: calls += 1 results.append({"type": "tool_result", "tool_use_id": u.id, "content": core.call_tool(u.name, u.input)}) messages.append({"role": "user", "content": results}) return { "mode": "loop", "model": model, "stop": stop, "model_turns": len(turns), "tool_calls": calls, "input_tokens": total_in, "output_tokens": total_out, "tokens_read": total_in, "seconds": round(time.time() - started, 1), "per_turn": turns, "final_text": final_text, } if __name__ == "__main__": ap = argparse.ArgumentParser() ap.add_argument("--model", default="claude-haiku-5-5") ap.add_argument("--max-turns", type=int, default=60) ap.add_argument("--tag", default="loop") args = ap.parse_args() reset_outputs() out = run(EXHIBIT_TASK, args.model, args.max_turns) out["verify"] = verify() results = core.ROOT / "results" results.mkdir(exist_ok=True) (results / f"result_{args.tag}.json").write_text(json.dumps(out, indent=2)) summary = {k: v for k, v in out.items() if k not in ("per_turn", "final_text")} print(json.dumps(summary, indent=2)) print("\nFINAL:\n" + out["final_text"])
  3. Run it. This makes real API calls and spends real money. At this size it is small, but look at your provider's usage page after the first run.
    python loop_mode.py --tag loop_run1
Check: expect roughly nine or ten model turns, 50 or more tool calls, and input tokens in the low hundreds of thousands. Compare calls to turns: that gap is the model batching. The check will probably report ok: false because fewer than twelve distinct kinds of object made it into the exhibit. If your run passes, that is fine; run it twice more before drawing a conclusion.
  1. Create Dockerfile. It installs two libraries and creates a non-root user. There are no secrets in it.
    # Network-ON sandbox for model-written code that calls free public APIs. # Contains no secrets. Compare with Lab 10's sandbox, which runs with --network none. FROM python:3.12-slim RUN pip install --no-cache-dir requests==2.32.3 pillow==10.4.0 \ && useradd --uid 1000 --create-home runner USER runner WORKDIR /app
  2. Create sandbox.py. The script arrives on stdin and is never written to disk. Read the docker run flags one at a time; each one removes something the script does not need. The timeout is enforced here, outside the container.
    """Run a model-written Python script in a locked-down container that HAS network access. Lab 10's sandbox uses --network none because it runs code that should never reach out. Here the tools must call a public API, so the network stays on and the protection comes from what the box does NOT contain: no secrets, no home directory, no write access outside one data folder, and hard limits enforced from outside the container. """ import subprocess import uuid import core IMAGE = "toolscale-sandbox" TIMEOUT_SECONDS = 120 MAX_OUTPUT_CHARS = 6000 def build_image() -> None: subprocess.run(["docker", "build", "-t", IMAGE, str(core.ROOT)], check=True) def run_script(code: str) -> str: """Execute `code` (sent on stdin, never written to disk) and return its combined output.""" name = f"toollab-{uuid.uuid4().hex[:8]}" cmd = [ "docker", "run", "--rm", "-i", "--name", name, "--read-only", "--tmpfs", "/tmp:size=64m", # nothing writable except /tmp and /data "--cap-drop", "ALL", "--security-opt", "no-new-privileges", "--pids-limit", "128", "--memory", "512m", "--cpus", "1", "-v", f"{core.ROOT}:/app:ro", # the tools, read-only "-v", f"{core.DATA_DIR}/cache:/data/cache", # shared API cache "-v", f"{core.DATA_DIR}/out:/data/out", # where results land "-e", "TOOLSCALE_DATA=/data", # the ONLY environment variable passed in IMAGE, "python", "-I", "-c", "import sys; sys.path.insert(0, '/app'); exec(sys.stdin.read())", ] try: proc = subprocess.run(cmd, input=code, capture_output=True, text=True, timeout=TIMEOUT_SECONDS) except subprocess.TimeoutExpired: # The timeout is enforced out here: code inside the box cannot disable it. subprocess.run(["docker", "rm", "-f", name], capture_output=True) return f"ERROR: script exceeded {TIMEOUT_SECONDS}s and was killed." out = (proc.stdout + proc.stderr).strip() or "(no output)" if proc.returncode != 0: out = f"EXIT CODE {proc.returncode}\n{out}" return out[-MAX_OUTPUT_CHARS:]
  3. Create probe.py: eight scripts that each try one thing. Build the image and run them.
    """Eight probes that show what the network-on sandbox does and does not allow.""" import sandbox PROBES = { "network works (public API)": "import urllib.request;print(urllib.request.urlopen('https://collectionapi.metmuseum.org/public/collection/v1/departments', timeout=10).status)", "tools import and make a real call": "from met_tools import *\nprint(get_object(45734)['title'])", "write to /app (should FAIL)": "open('/app/x', 'w').write('hi')", "write to /data/out (should work)": "open('/data/out/probe.txt', 'w').write('ok');print('wrote')", "see your home folder (should FAIL)": "import os;print(os.listdir('/Users'))", "environment variables": "import os;print(sorted(os.environ))", "who am I": "import os;print('uid', os.getuid())", "infinite loop (should be killed)": "while True: pass", } sandbox.TIMEOUT_SECONDS = 8 # keep the last probe short for name, code in PROBES.items(): print(f"\n## {name}\n" + sandbox.run_script(code).splitlines()[-1][:200])
  4. python -c "import sandbox; sandbox.build_image()" python probe.py
Check: the network probe prints 200 and the tools import and make a real call. Writing to /app fails with a read-only error, writing to /data/out works, /Users does not exist, and you are user 1000. The environment lists image defaults plus one variable, and none of your API keys. The infinite loop is killed after 8 seconds. If you use Colima on a Mac, start it with the lab folder mounted writable, for example colima start --mount /path/to/your/lab:w; Docker Desktop shares your user folder by default. This was verified on macOS with Colima. Docker on Linux should behave the same, but bind-mount file ownership can differ.
  1. Create code_mode.py. The system prompt is generated from the same docstrings the loop version sent as JSON schemas, so both modes see identical contracts. The model gets one tool, run_python, and a script that raises an error comes back to it as text so it can fix and rerun.
    """Code mode: the model writes one script that calls the same tools as Python functions. python code_mode.py [--model claude-haiku-5-5] [--max-turns 8] The intermediate data (hundreds of object records) stays inside the sandbox. Only what the script prints comes back to the model. """ import argparse import inspect import json import time import anthropic import core import met_tools # noqa: F401 (registers the tools) import sandbox from tasks import EXHIBIT_TASK, reset_outputs, verify def api_docs() -> str: """The same contract the loop version sent as JSON schemas, written as a Python API.""" parts = [] for fn in core.TOOLS.values(): parts.append(f"def {fn.__name__}{inspect.signature(fn)}:\n \"\"\"{inspect.getdoc(fn)}\"\"\"") return "\n\n".join(parts) SYSTEM = f"""You are a museum curator's assistant. You complete tasks by writing ONE Python script and running it with the run_python tool. Import the tools like this: from met_tools import * (everything below is then available) {api_docs()} Rules: - The script runs in a sandbox. Tool functions return plain Python data (dicts and lists). - Do the whole task in a single script: loop, filter and check requirements in code. Do not print raw records; print only short summaries and the final answer. - If a tool raises an error, the traceback comes back to you. Fix the script and run it again. - ToolError means a tool refused the request; read its message. """ RUN_PYTHON = { "name": "run_python", "description": "Run a Python script in the sandbox and return everything it printed.", "input_schema": {"type": "object", "properties": {"code": {"type": "string", "description": "The full script."}}, "required": ["code"]}, } def run(task: str, model: str, max_turns: int) -> dict: client = anthropic.Anthropic() messages = [{"role": "user", "content": task}] turns, scripts, total_in, total_out = [], 0, 0, 0 started = time.time() final_text, stop = "", "max_turns" for n in range(1, max_turns + 1): resp = client.messages.create( model=model, max_tokens=8000, system=SYSTEM, tools=[RUN_PYTHON], messages=messages, thinking={"type": "disabled"}, ) total_in += resp.usage.input_tokens total_out += resp.usage.output_tokens uses = [b for b in resp.content if b.type == "tool_use"] turns.append({"turn": n, "input_tokens": resp.usage.input_tokens, "output_tokens": resp.usage.output_tokens, "scripts": len(uses)}) messages.append({"role": "assistant", "content": resp.content}) if not uses: final_text = "".join(b.text for b in resp.content if b.type == "text") stop = resp.stop_reason break results = [] for u in uses: scripts += 1 results.append({"type": "tool_result", "tool_use_id": u.id, "content": sandbox.run_script(u.input["code"])}) messages.append({"role": "user", "content": results}) return { "mode": "code", "model": model, "stop": stop, "model_turns": len(turns), "scripts_run": scripts, "input_tokens": total_in, "output_tokens": total_out, "tokens_read": total_in, "seconds": round(time.time() - started, 1), "per_turn": turns, "final_text": final_text, } if __name__ == "__main__": ap = argparse.ArgumentParser() ap.add_argument("--model", default="claude-haiku-5-5") ap.add_argument("--max-turns", type=int, default=8) ap.add_argument("--tag", default="code") ap.add_argument("--build", action="store_true", help="build the sandbox image first") args = ap.parse_args() if args.build: sandbox.build_image() reset_outputs() out = run(EXHIBIT_TASK, args.model, args.max_turns) out["verify"] = verify() results = core.ROOT / "results" results.mkdir(exist_ok=True) (results / f"result_{args.tag}.json").write_text(json.dumps(out, indent=2)) print(json.dumps({k: v for k, v in out.items() if k not in ("per_turn", "final_text")}, indent=2)) print("\nFINAL:\n" + out["final_text"])
  2. Run it three times. Then run the loop three times if you have not already, so you have three of each.
    python code_mode.py --tag code_run1 python code_mode.py --tag code_run2 python code_mode.py --tag code_run3
  3. Create report.py and run it. It compares every saved run, including the input tokens on each turn.
    """Print a comparison of every saved run: python report.py""" import json import statistics from pathlib import Path runs = [] for path in sorted(Path(__file__).parent.glob("results/result_*.json")): r = json.loads(path.read_text()) r["name"] = path.stem.removeprefix("result_") runs.append(r) print(f"{'run':12}{'mode':6}{'turns':>6}{'calls':>7}{'input tokens':>14}{'passed':>8}") for r in runs: calls = r.get("tool_calls", r.get("scripts_run")) print(f"{r['name']:12}{r['mode']:6}{r['model_turns']:>6}{calls:>7}{r['input_tokens']:>14,}{str(r['verify']['ok']):>8}") print() for mode in ("loop", "code"): tokens = [r["input_tokens"] for r in runs if r["mode"] == mode] if tokens: print(f"{mode}: median {int(statistics.median(tokens)):,} input tokens " f"(range {min(tokens):,} to {max(tokens):,}, {len(tokens)} runs)") print("\ninput tokens on each turn (the model re-reads everything so far):") for r in runs: print(f"{r['name']:12}" + " ".join(f"{t['input_tokens']:>7,}" for t in r["per_turn"]))
  4. python report.py
Check: the code runs read far fewer input tokens than the loop runs, and the per-turn lines show why: the loop's numbers climb turn after turn, while the code run stays low. Expect the code runs to differ from each other, and for at least one to need more than one script. Expect the check to pass for code mode. If a code run is slower on the clock than a loop run, that is real: container start-up and retries count.
  1. Call the old, retired search URL through your own fetch and read what comes back.
    python -c "import core, met_tools as m try: core.fetch(m.BASE + '/v1/search', {'q': 'cat'}) except core.ToolError as e: print(e)"
  2. Now imagine fetch had caught that error and returned an empty dict, and search_collection had carried on. Write down, in two or three sentences, what the loop version would have done next, and what the code version would have printed. Which of the two would have told you something was wrong?
Check: at the time of writing the API answers HTTP 410 and says the endpoint was retired on 2026-10-01 and names /v1.1/search as the replacement. If the Met has changed this again, you will see whatever it says now, which is the point: the error text is the only documentation you get at that moment.
Watch-out

These labs make real model calls and real requests to a public museum's servers. The cache folder means reruns are free and gentle, so delete cache/ only when you want fresh data. Keep the honest User-Agent in core.py: if an API ever blocks you, do not disguise your client to get around it. Read the error, slow down, or pick another API.

Project

Project · Bring your own example

The museum curator is one example of a shape: a free API, a fan-out of calls, big records you mostly do not need, a requirement a script can check exactly, and something fun to open at the end. A Pokemon team builder is another. This project has you find a third and prove the lesson holds on it.

  1. Paste this prompt into any capable model. It asks for five ideas, recommends one, and tells you how to test each API before you build anything.
    I'm building a hands-on lesson that compares two ways of letting an AI model call tools: one tool call per step, versus the model writing one script that calls the tools as Python functions. I need a new example task to teach it with. A good example has ALL of these: 1. It uses a free public web API that needs no account or API key. Name the exact base URL and the 2 to 4 endpoints involved. 2. It needs at least 10 tool calls to finish, including a fan-out: one search returns many ids, and each id needs its own detail call. 3. Each detail call returns a large JSON record (a couple of KB or more), most of which the task does not need. 4. It has a checkable requirement that a script can verify exactly, such as "no two items of the same type", "total cost under a budget", or "at least 4 different categories". State the check in one line of code. 5. It ends in something a person would enjoy opening: an image grid, a web page, a printable card, a playlist. 6. It is safe: read-only calls, nothing that posts, buys or messages anyone. 7. All 5 ideas must differ from each other and from the two reference examples in subject and in API. Give exactly one API per idea, with plain GET requests: no "alternatively", no POST, no key. Two reference examples of the right shape: - A museum exhibit curator: search a museum's open collection, fetch each object, keep only public-domain pieces from at least 4 departments with no two of the same kind, then build a gallery page. - A Pokemon team builder: fetch many creature records, then pick six whose type weaknesses do not overlap. Give me 5 NEW ideas in a table with these columns: name, the API and endpoints, the task in one sentence, the checkable requirement, the final artifact, and why the data is large. Then recommend one. For the one you recommend, also give me: - the 5 to 7 tool functions as Python signatures with one-line docstrings, - the full task prompt I would give the model, with the checkable requirements spelled out, - the verify() rules. Be honest about what you cannot know: you cannot check whether an API is still online or still free today. For each API, give me a one-line curl command I can run to test it before I build anything, and tell me what a healthy response looks like.
  2. Run the curl test it gives you for the API you pick. Free APIs disappear, move behind a key, or change shape, and the model cannot check which today. If the test fails, pick another idea; that is a normal outcome and a cheap one.
  3. Write your_tools.py beside met_tools.py. It imports fetch, tool and ToolError from core.py and changes nothing in it. Write the docstrings for the model.
  4. Write your own task prompt and verify() in a copy of tasks.py, from the requirements the model proposed. Make at least one requirement the loop is likely to get slightly wrong, such as a count of distinct values.
  5. Run loop mode three times and code mode three times. Save the results and run report.py.
  6. Write five lines: your medians for both modes, whether each passed the check, one thing that surprised you, and when you would choose each mode for a real product.

Done when:

  • Your tools reuse core.py without edits to it.
  • You tested the API with curl before building on it.
  • Both modes were run three times on the same task, and the check decides pass or fail, not the model's own summary.
  • Your five lines say what you measured and where it might not generalize, in your own words.

Sources

  • Anthropic Engineering (2025). Code execution with MCP. Published 4 November 2025. The 150,000 to 2,000 token example, progressive disclosure of tools, filtering in code, and the sandboxing costs. It also notes that Cloudflare describes a similar approach as "Code Mode".
  • The Metropolitan Museum of Art. Collection API documentation. Images and data used here are the museum's public-domain open-access records.
  • All token counts, turn counts and pass results in this lesson come from six runs made while writing it (three per mode, Claude Haiku 5.5, prompt caching off, one task). They are measurements of this task, not benchmarks.
← Back to the course map