Real tool-use agent with native function calling#
Track 08 - Production · Notebook 02 · Runtime: ~60s LIVE, <1s replay
The same 20 tasks as 04_agents/01_react_from_scratch, but the agent is the actual Anthropic API doing native tool_use. No regex parser, no Action: name[arg] grammar — the model emits structured tool calls and the SDK gives them back as objects.
from llm_systems_cookbook.nb import bootstrap
from llm_systems_cookbook._utils import repo_root
import ast
import json
import operator as op
import os
import time
from datetime import datetime, timedelta, timezone
from pathlib import Path
s = bootstrap("08_production_03_tool_use_agent")
LIVE = bool(os.environ.get("ANTHROPIC_API_KEY"))
MODEL = os.environ.get("MODEL_ANTHROPIC", "claude-sonnet-4-6")
FIXTURE = json.loads((Path(repo_root()) / "notebooks/08_production/_fixtures/03_tool_use.json").read_text())
TASKS = FIXTURE["tasks"]
print(f"mode={'LIVE' if LIVE else 'REPLAY'} tasks={len(TASKS)} model={MODEL}")
Tools — Python functions plus a JSON schema#
Each tool is a regular Python function. The model sees the JSON schema (name, description, input_schema) at request time; the SDK returns a structured tool_use block with name and input we unpack and dispatch.
_OPS = {ast.Add: op.add, ast.Sub: op.sub, ast.Mult: op.mul, ast.Div: op.truediv,
ast.Pow: op.pow, ast.USub: op.neg, ast.UAdd: op.pos, ast.Mod: op.mod}
def _eval(node):
if isinstance(node, ast.Constant): return node.value
if isinstance(node, ast.BinOp): return _OPS[type(node.op)](_eval(node.left), _eval(node.right))
if isinstance(node, ast.UnaryOp): return _OPS[type(node.op)](_eval(node.operand))
raise ValueError(type(node).__name__)
def calculator(expression: str) -> str:
v = _eval(ast.parse(expression, mode="eval").body)
return str(int(v)) if isinstance(v, float) and v.is_integer() else str(v)
FACTS = {
"capital of france": "Paris", "capital of japan": "Tokyo",
"tallest mountain": "Mount Everest",
"author of hamlet": "William Shakespeare",
"chemical symbol for gold": "Au",
"speed of light": "299,792,458 m/s in vacuum",
"painter of the mona lisa": "Leonardo da Vinci",
"number of continents": "Seven",
}
def wiki_lookup(query: str) -> str:
q = query.lower().strip().rstrip("?")
if q in FACTS: return FACTS[q]
for k, v in FACTS.items():
if k in q or q in k: return v
return "no match"
NOW = datetime(2026, 4, 17, 12, 0, tzinfo=timezone.utc)
def get_date(spec: str) -> str:
s = spec.lower().strip()
return {
"year": str(NOW.year), "month": NOW.strftime("%B"), "day": NOW.strftime("%A"),
"today": NOW.date().isoformat(),
"tomorrow": (NOW + timedelta(days=1)).date().isoformat(),
"yesterday": (NOW - timedelta(days=1)).date().isoformat(),
}.get(s, f"unknown spec {spec!r}")
TOOLS = {"calculator": calculator, "wiki_lookup": wiki_lookup, "get_date": get_date}
TOOL_SCHEMAS = [
{"name": "calculator", "description": "Evaluate an arithmetic expression.",
"input_schema": {"type": "object", "required": ["expression"],
"properties": {"expression": {"type": "string"}}}},
{"name": "wiki_lookup", "description": "Look up a short fact.",
"input_schema": {"type": "object", "required": ["query"],
"properties": {"query": {"type": "string"}}}},
{"name": "get_date", "description": "Get today, tomorrow, year, month, etc.",
"input_schema": {"type": "object", "required": ["spec"],
"properties": {"spec": {"type": "string",
"enum": ["today","tomorrow","yesterday","year","month","day"]}}}},
]
The agent loop#
Three-state machine: send messages → if stop_reason == "tool_use", run the tools, append their results, loop; otherwise emit the final text and stop. ~25 lines of real code.
def agent_run(task: str, fixture_idx: int) -> dict:
if not LIVE:
rec = FIXTURE["native_tool_use"][fixture_idx]
# Tool functions are deterministic; run them locally to verify.
tool_result = TOOLS[rec["tool"]](**rec["tool_input"])
return {**rec, "task": task, "tool_result": tool_result}
import anthropic # noqa: PLC0415
client = anthropic.Anthropic()
messages: list = [{"role": "user", "content": task}]
n_tool_calls = 0
used_tool, used_input, tool_result = None, None, None
t0 = time.perf_counter()
while True:
resp = client.messages.create(
model=MODEL, max_tokens=512, tools=TOOL_SCHEMAS, messages=messages,
)
if resp.stop_reason == "end_turn":
text = "".join(b.text for b in resp.content if b.type == "text")
return {"task": task, "final": text, "tool": used_tool,
"tool_input": used_input, "tool_result": tool_result,
"n_tool_calls": n_tool_calls,
"input_tokens": resp.usage.input_tokens,
"output_tokens": resp.usage.output_tokens,
"cost_usd": (resp.usage.input_tokens * 3 + resp.usage.output_tokens * 15) / 1e6,
"latency_s": time.perf_counter() - t0}
if resp.stop_reason != "tool_use":
raise RuntimeError(f"unexpected stop_reason {resp.stop_reason}")
messages.append({"role": "assistant", "content": resp.content})
results = []
for block in resp.content:
if block.type != "tool_use":
continue
n_tool_calls += 1
used_tool, used_input = block.name, block.input
try:
tool_result = TOOLS[block.name](**block.input)
except Exception as e: # noqa: BLE001 — surface to model
tool_result = f"ERROR: {e}"
results.append({"type": "tool_result", "tool_use_id": block.id,
"content": tool_result})
messages.append({"role": "user", "content": results})
Run all 20 tasks#
native = [agent_run(t, i) for i, (t, _expected) in enumerate(TASKS)]
def graded(rec, expected): return expected.lower() in rec["final"].lower()
native_correct = sum(graded(r, e) for r, (_, e) in zip(native, TASKS))
native_cost = sum(r["cost_usd"] for r in native)
native_lat = sum(r["latency_s"] for r in native)
print(f"native tool_use {native_correct}/{len(TASKS)} correct "
f"${native_cost:.4f} {native_lat:.1f}s")
for r, (_, e) in zip(native[:4], TASKS[:4]):
print(f" [{r['tool']:>11}] {r['task'][:36]:<36} → {r['final'][:50]}")
Compare to prompted ReAct + regex parser#
Same model, same 20 tasks, but this time tools are described in the system prompt and we regex-parse Action: name[arg]. The fixture records what actually happens: most calls succeed cheaply, but two tasks (#7 and #18) blow their step budget after the model emits a Thought: without a follow-up Action:.
react = FIXTURE["prompted_react"]
react_correct = sum(r["success"] for r in react)
react_parse_errors = sum(r["parse_errors"] for r in react)
react_cost = sum(r["cost_usd"] for r in react)
react_lat = sum(r["latency_s"] for r in react)
print(f"prompted ReAct {react_correct}/{len(TASKS)} correct "
f"parse errors {react_parse_errors} "
f"${react_cost:.4f} {react_lat:.1f}s")
print()
print(f"native is {react_cost / native_cost:.1f}x cheaper and "
f"{react_lat / native_lat:.1f}x faster, with 0 parse errors.")
Visualisation#
import matplotlib.pyplot as plt
fig, axes = plt.subplots(1, 3, figsize=(11, 3.2))
labels = ["native\ntool_use", "prompted\nReAct"]
colors = ["tab:green", "tab:orange"]
axes[0].bar(labels, [native_correct, react_correct], color=colors)
axes[0].set_ylabel("tasks correct (of 20)"); axes[0].set_ylim(0, 21)
axes[0].set_title("success")
for i, v in enumerate([native_correct, react_correct]):
axes[0].text(i, v + 0.3, str(v), ha="center")
axes[1].bar(labels, [0, react_parse_errors], color=colors)
axes[1].set_ylabel("parse errors"); axes[1].set_title("format breakage")
for i, v in enumerate([0, react_parse_errors]):
axes[1].text(i, v + 0.05, str(v), ha="center")
axes[2].bar(labels, [native_cost * 100, react_cost * 100], color=colors)
axes[2].set_ylabel("total cost (cents)"); axes[2].set_title(f"{len(TASKS)}-task cost")
for i, v in enumerate([native_cost * 100, react_cost * 100]):
axes[2].text(i, v + 0.02, f"{v:.2f}c", ha="center")
fig.tight_layout(); plt.show()
Checks#
s.check(
"all_tasks_solved_with_native_tool_use",
lambda: native_correct == len(TASKS),
msg=f"{native_correct}/{len(TASKS)}",
)
s.check(
"exactly_one_tool_call_per_task",
lambda: all(r["n_tool_calls"] == 1 for r in native),
msg=f"tool_calls = {[r['n_tool_calls'] for r in native]}",
)
s.check(
"tools_dispatched_correctly",
lambda: all(r["tool"] in TOOLS for r in native),
msg=f"tools used = {sorted({r['tool'] for r in native})}",
)
s.check(
"native_outperforms_prompted_react_on_success",
lambda: native_correct >= react_correct + 1,
msg=f"native={native_correct}/{len(TASKS)} react={react_correct}/{len(TASKS)}",
)
s.check(
"native_at_least_40pct_cheaper",
lambda: native_cost <= react_cost * 0.6,
msg=f"native=${native_cost:.4f} react=${react_cost:.4f} "
f"savings={1 - native_cost / react_cost:.0%}",
)
Notes for production#
Parallel tool calls.
disable_parallel_tool_usedefaults toFalse; for tasks that have independent sub-questions (“look up X and compute Y”) the model returns multipletool_useblocks in one turn. Run them concurrently; append all the results in one user turn.Cache the tool definitions. Put
cache_control={"type":"ephemeral"}on the last tool. The fulltoolsarray is part of the cacheable prefix and bills at cache rates from the second call onward.Tool choice forcing.
tool_choice={"type": "tool", "name": "..."}forces a specific tool;{"type": "any"}forces some tool;{"type": "auto"}is the default. Useful for routing nodes in an agent graph.Validate inputs. The
input_schemaconstrains structure but not semantics. Acalculatortool should still reject__import__expressions; aread_filetool should still resolve paths and reject anything outside the workspace.Computer use (Sonnet 4+): the same protocol with
screenshot,mouse_*,key_*tools turns the model into a browser/OS agent. Same loop, sametool_usemachinery — just bigger tool surface.
s.summary()
s.save()