Build your first tool-using agent (offline, in pure Python)

A customer asks a support assistant: “It's Monday. When will orders A100 and A200 arrive?” Answering takes three tool calls, two of which can run at once, and one that depends on what the first returned. That is the smallest problem that exercises everything an agent loop has to get right, and it is what we build here: first offline with a scripted model, then against the real Claude API.

What real APIs actually send

Many first agent tutorials model a tool call as {"name": ..., "args": ...} and claim you can “swap one function” to go live. You cannot, for two reasons. Real APIs link every result to its call by id, and they can request several tools in one response. A loop that assumes one call per turn, or appends results without ids, will be rejected by the API or will silently mis-pair results the first time the model parallelises.

texttext17 lines

# Anthropic Messages API: one assistant turn may ask for several tools at once.
{"role": "assistant", "content": [
  {"type": "text", "text": "I'll look up both shipments."},
  {"type": "tool_use", "id": "toolu_01", "name": "lookup_shipment", "input": {"order_id": "A100"}},
  {"type": "tool_use", "id": "toolu_02", "name": "lookup_shipment", "input": {"order_id": "A200"}}
]}
# Your reply: ONE user-role message, one tool_result per tool_use, matched by id.
{"role": "user", "content": [
  {"type": "tool_result", "tool_use_id": "toolu_01", "content": "{\"status\": \"in_transit\", ...}"},
  {"type": "tool_result", "tool_use_id": "toolu_02", "content": "{\"status\": \"awaiting_pickup\", ...}"}
]}

# OpenAI Chat Completions: the same idea, different field names.
{"role": "assistant", "tool_calls": [{"id": "call_a", "type": "function",
  "function": {"name": "lookup_shipment", "arguments": "{\"order_id\": \"A100\"}"}}, ...]}
{"role": "tool", "tool_call_id": "call_a", "content": "{...}"}   # one tool message per call
# (OpenAI's Responses API uses function_call / function_call_output items linked by call_id.)

The rules for the Anthropic shape, which the offline code below follows exactly: append the assistant's whole content list unchanged; reply with one user-role message containing a tool_result for every tool_use, each carrying the matching tool_use_id; set is_error: true on failed calls instead of dropping them. OpenAI's Chat Completions uses one role: "tool" message per call with tool_call_id; the pairing rule is the same.

Your loop

Model

Tools

user questionYour loop → Model

tool_use ×2 (01, 02)Model → Your loop

run bothYour loop → Tools

tool_result 01, 02Your loop → Model

tool_use 03Model → Your loop

run add_daysYour loop → Tools

tool_result 03Your loop → Model

end_turn + answerModel → Your loop

Three model calls, three tool executions. The model never runs anything itself.

The offline program

Save as agent.py and run python3 agent.py (Python 3.10+, no installs). scripted_model returns responses in the Messages API shape, as plain dicts, with fixed decisions so the run is repeatable.

example.pypython100 lines

"""A tool-using agent loop, offline. Messages use the Anthropic Messages API shape."""
from __future__ import annotations

import inspect
import json
from typing import Any, Callable

# ---- Tools: plain functions the agent is allowed to call ----
SHIPMENTS = {"A100": {"status": "in_transit", "days_left": 2},
             "A200": {"status": "awaiting_pickup", "days_left": None}}
WEEKDAYS = ["Mon", "Tue", "Wed", "Thu", "Fri", "Sat", "Sun"]


def lookup_shipment(order_id: str) -> dict[str, Any]:
    shipment = SHIPMENTS.get(order_id)
    if shipment is None:
        return {"error": f"no shipment for order {order_id}"}
    return {"order_id": order_id, **shipment}


def add_days(start_day: str, days: int) -> dict[str, Any]:
    if start_day not in WEEKDAYS:
        return {"error": f"start_day must be one of {WEEKDAYS}"}
    return {"day": WEEKDAYS[(WEEKDAYS.index(start_day) + days) % 7]}


TOOLS: dict[str, Callable[..., dict[str, Any]]] = {
    "lookup_shipment": lookup_shipment,
    "add_days": add_days,
}


def run_tool(name: str, args: dict[str, Any]) -> tuple[dict[str, Any], bool]:
    """Returns (result, is_error). Bad calls from the model become errors it can read.
    Exceptions raised *inside* a tool are bugs in our code: they propagate."""
    fn = TOOLS.get(name)
    if fn is None:
        return {"error": f"unknown tool {name}"}, True
    try:
        inspect.signature(fn).bind(**args)  # wrong or missing argument names
    except TypeError as exc:
        return {"error": f"bad arguments for {name}: {exc}"}, True
    result = fn(**args)
    return result, "error" in result


# ---- STAND-IN for a real model: same response shape, scripted decisions ----
def scripted_model(messages: list[dict[str, Any]]) -> dict[str, Any]:
    results = [block for m in messages if m["role"] == "user" and isinstance(m["content"], list)
               for block in m["content"] if block["type"] == "tool_result"]
    if not results:  # turn 1: two independent lookups, requested in ONE response
        return {"stop_reason": "tool_use", "content": [
            {"type": "text", "text": "I'll look up both shipments."},
            {"type": "tool_use", "id": "toolu_01", "name": "lookup_shipment", "input": {"order_id": "A100"}},
            {"type": "tool_use", "id": "toolu_02", "name": "lookup_shipment", "input": {"order_id": "A200"}},
        ]}
    if len(results) == 2:  # turn 2: depends on what turn 1 observed
        a100 = json.loads(next(r["content"] for r in results if r["tool_use_id"] == "toolu_01"))
        return {"stop_reason": "tool_use", "content": [
            {"type": "tool_use", "id": "toolu_03", "name": "add_days",
             "input": {"start_day": "Mon", "days": a100["days_left"]}},
        ]}
    day = json.loads(next(r["content"] for r in results if r["tool_use_id"] == "toolu_03"))["day"]
    return {"stop_reason": "end_turn", "content": [
        {"type": "text", "text": f"A100 is in transit and should arrive {day}. "
                                 "A200 has not been picked up yet, so it has no date."},
    ]}


def run_agent(question: str, model: Callable[[list[dict[str, Any]]], dict[str, Any]],
              max_steps: int = 5) -> str:
    messages: list[dict[str, Any]] = [{"role": "user", "content": question}]
    for step in range(1, max_steps + 1):
        response = model(messages)                                        # THINK
        messages.append({"role": "assistant", "content": response["content"]})
        if response["stop_reason"] == "end_turn":                         # STOP: model is done
            print(f"step {step}: final answer")
            return "".join(b["text"] for b in response["content"] if b["type"] == "text")
        if response["stop_reason"] != "tool_use":                         # max_tokens, refusal, ...
            return f"Stopped: unexpected stop_reason {response['stop_reason']!r}."
        results = []
        for block in response["content"]:
            if block["type"] != "tool_use":
                continue
            result, is_error = run_tool(block["name"], block["input"])    # ACT
            print(f"step {step}: {block['id']} {block['name']}({block['input']}) -> {result}")
            results.append({"type": "tool_result", "tool_use_id": block["id"],
                            "content": json.dumps(result), "is_error": is_error})
        messages.append({"role": "user", "content": results})             # OBSERVE: all results, one message
    return f"Stopped: step budget of {max_steps} used up before the task finished."


def main() -> None:
    question = "It's Monday. When will orders A100 and A200 arrive?"
    print("answer:", run_agent(question, scripted_model))
    print("answer:", run_agent(question, scripted_model, max_steps=2))


if __name__ == "__main__":
    main()

Output:

texttext9 lines

step 1: toolu_01 lookup_shipment({'order_id': 'A100'}) -> {'order_id': 'A100', 'status': 'in_transit', 'days_left': 2}
step 1: toolu_02 lookup_shipment({'order_id': 'A200'}) -> {'order_id': 'A200', 'status': 'awaiting_pickup', 'days_left': None}
step 2: toolu_03 add_days({'start_day': 'Mon', 'days': 2}) -> {'day': 'Wed'}
step 3: final answer
answer: A100 is in transit and should arrive Wed. A200 has not been picked up yet, so it has no date.
step 1: toolu_01 lookup_shipment({'order_id': 'A100'}) -> {'order_id': 'A100', 'status': 'in_transit', 'days_left': 2}
step 1: toolu_02 lookup_shipment({'order_id': 'A200'}) -> {'order_id': 'A200', 'status': 'awaiting_pickup', 'days_left': None}
step 2: toolu_03 add_days({'start_day': 'Mon', 'days': 2}) -> {'day': 'Wed'}
answer: Stopped: step budget of 2 used up before the task finished.

Tools: structured results, readable errors

Each tool returns a dict, never prose, so the model reasons over clean fields. Expected failures (an unknown order id, a bad weekday) come back as an error field rather than an exception: the model can read “no shipment for order A999” and ask the user to check the number. Shaping inputs, outputs and errors for a model reader is tool design, and it moves agent quality as much as the prompt does.

Validation that does not swallow bugs

run_tool() separates two kinds of failure that a blanket except TypeError would merge. The model's mistakes (an unknown tool name, a misspelt or missing argument) are caught by checking the registry and by inspect.signature(fn).bind(**args), which raises TypeError only for argument mismatches, before the tool runs. They return is_error: True so the model can correct itself. Your bugs (a KeyError inside a tool, or a TypeError raised deep in its logic) are not caught at all. They crash the run and show up in your error tracker, which is where a bug belongs. Wrapping the call itself in except TypeError would hand your stack trace to the model as if it had made a typo.

bind() checks names, not types: {"days": "two"} passes it. On the real API, strict: true on the tool definition makes the service guarantee schema-valid input; offline, add a schema check if your tools are not defensive.

The loop and its exits

run_agent() is the agent loop: think (call the model), act (run every tool_use block), observe (append all results as one message). It has three exits, all deliberate: end_turn, the happy path; any other stop_reason, such as max_tokens (the response was cut off, so its tool input may be incomplete) or refusal, which ends the run with a reason instead of guessing; and the step budget.

The second run in the output shows the budget at work. With max_steps=2 the agent spends both steps on tools and stops with an explicit message rather than looping or pretending to have finished. Set the budget from the task's expected depth plus headroom (this task needs 3; 5 is reasonable), and alert when the hit rate passes ~1–2% of runs, because that usually means a tool keeps failing in a way the model cannot fix. These are your stop conditions.

Tests for the offline loop

Scripted models make the loop testable without a network. These tests pin the trajectory, the id pairing rule, both validation paths, the bug-propagation guarantee and the budget. Save next to agent.py; run pytest -q, or python3 test_agent.py with no installs.

example.pypython70 lines

# test_agent.py: run with "pytest -q" next to agent.py (or "python3 test_agent.py").
from typing import Any

import agent
from agent import run_agent, run_tool, scripted_model


def test_trajectory_is_two_parallel_lookups_then_one_add_days() -> None:
    calls: list[str] = []

    def spy(messages: list[dict[str, Any]]) -> dict[str, Any]:
        response = scripted_model(messages)
        calls.append(",".join(b["name"] for b in response["content"] if b["type"] == "tool_use"))
        return response

    answer = run_agent("It's Monday. When will orders A100 and A200 arrive?", spy)
    assert calls == ["lookup_shipment,lookup_shipment", "add_days", ""]
    assert "Wed" in answer


def test_every_tool_use_gets_a_matching_tool_result() -> None:
    seen: list[list[dict[str, Any]]] = []

    def spy(messages: list[dict[str, Any]]) -> dict[str, Any]:
        seen.append(messages)
        return scripted_model(messages)

    run_agent("It's Monday. When will orders A100 and A200 arrive?", spy)
    history = seen[-1]
    for i, msg in enumerate(history):
        if msg["role"] == "assistant":
            asked = {b["id"] for b in msg["content"] if b["type"] == "tool_use"}
            answered = {b["tool_use_id"] for b in history[i + 1]["content"]} if asked else set()
            assert asked == answered


def test_bad_arguments_become_a_readable_error() -> None:
    result, is_error = run_tool("add_days", {"start": "Mon", "days": 2})
    assert is_error and "bad arguments" in result["error"]


def test_unknown_tool_is_refused() -> None:
    result, is_error = run_tool("delete_order", {"order_id": "A100"})
    assert is_error and "unknown tool" in result["error"]


def test_bugs_inside_tools_are_not_swallowed() -> None:
    def broken(order_id: str) -> dict[str, Any]:
        raise KeyError("carrier")  # a bug in our code, not the model's mistake

    agent.TOOLS["broken"] = broken
    try:
        run_tool("broken", {"order_id": "A100"})
    except KeyError:
        pass
    else:
        raise AssertionError("tool bug was swallowed")
    finally:
        del agent.TOOLS["broken"]


def test_budget_stops_visibly() -> None:
    assert run_agent("When will A100 arrive?", scripted_model, max_steps=1).startswith("Stopped")


if __name__ == "__main__":
    for name, fn in list(globals().items()):
        if name.startswith("test_"):
            fn()
            print("ok", name)

test_trajectory_is_two_parallel_lookups_then_one_add_days is a tiny trajectory eval: it checks the path, not just the final answer. With a real model you would assert looser properties (both lookups happen before add_days, no tool is called twice with the same input) because the exact path can vary between runs.

The same loop against Claude

This needs network access and an ANTHROPIC_API_KEY; it will not run offline. It reuses run_tool() from the offline file, so the registry, validation and bug-propagation rules are identical. What changes is real: tool definitions with JSON Schemas, SDK response objects instead of dicts, a timeout, and retries.

example.pypython58 lines

# agent_real.py: the same loop against the Claude API. Needs network and a key.
#   pip install "anthropic==1.13.0"
#   export ANTHROPIC_API_KEY=...
from __future__ import annotations

import json
from typing import Any

import anthropic

from agent import run_tool  # same tools, same validation as the offline version

CLIENT = anthropic.Anthropic(timeout=60.0, max_retries=3)  # SDK retries 408/409/429/5xx with backoff
MODEL = "claude-opus-5-5"

TOOL_SPECS = [
    {
        "name": "lookup_shipment",
        "description": "Get carrier status and days left in transit for one order id, e.g. A100.",
        "input_schema": {"type": "object", "properties": {"order_id": {"type": "string"}},
                         "required": ["order_id"], "additionalProperties": False},
        "strict": True,  # the API guarantees input matches this schema
    },
    {
        "name": "add_days",
        "description": "Return the weekday that is N days after start_day (Mon..Sun).",
        "input_schema": {"type": "object",
                         "properties": {"start_day": {"type": "string", "enum": ["Mon", "Tue", "Wed", "Thu", "Fri", "Sat", "Sun"]},
                                        "days": {"type": "integer"}},
                         "required": ["start_day", "days"], "additionalProperties": False},
        "strict": True,
    },
]


def run_agent(question: str, max_steps: int = 8) -> str:
    messages: list[dict[str, Any]] = [{"role": "user", "content": question}]
    for _ in range(max_steps):
        response = CLIENT.messages.create(model=MODEL, max_tokens=16000,
                                          tools=TOOL_SPECS, messages=messages)
        # Append the whole content list: tool_use blocks (and thinking blocks) must go back unchanged.
        messages.append({"role": "assistant", "content": response.content})
        if response.stop_reason == "end_turn":
            return "".join(b.text for b in response.content if b.type == "text")
        if response.stop_reason != "tool_use":  # max_tokens, refusal, pause_turn ...
            return f"Stopped: stop_reason={response.stop_reason}"
        results = []
        for block in response.content:
            if block.type == "tool_use":  # there may be several in one response
                result, is_error = run_tool(block.name, block.input)
                results.append({"type": "tool_result", "tool_use_id": block.id,
                                "content": json.dumps(result), "is_error": is_error})
        messages.append({"role": "user", "content": results})  # every result, ONE user message
    return f"Stopped: step budget of {max_steps} used up."


if __name__ == "__main__":
    print(run_agent("It's Monday. When will orders A100 and A200 arrive?"))

messages.append({"role": "assistant", "content": response.content}) sends back the model's content blocks untouched, including any thinking blocks, which current models require to be returned unchanged.
timeout=60.0, max_retries=3: the SDK retries 408/409/429/5xx with exponential backoff, so a rate-limit blip costs seconds, not a failed run. Worst-case wall clock per call is roughly timeout × (max_retries + 1); budget for it.
strict: True with additionalProperties: false makes argument shape a guarantee rather than a hope, which is why the offline bind() check rarely fires here.
Every non-tool_use, non-end_turn stop reason ends the run explicitly. Treating max_tokens as success is the most common silent bug in hand-written loops.

What this still leaves out

No persistence if the process dies mid-run (durable execution), no trimming of a history that grows every step (context engineering), no concurrent execution of parallel calls, and no per-tool rate limits. The SDK's tool runner (client.beta.messages.tool_runner) can own the loop for you once you understand what it is doing; this file is that understanding.

Key takeaways

The contract with a real API is ids and plurality: every call has an id, a turn can hold several calls, and every call gets exactly one result in one reply message.
Catch the model's mistakes before the tool runs and let your own exceptions crash; a loop that hides bugs as “tool errors” teaches the model to work around your defects.
A scripted model in the real response shape is what makes an agent unit-testable; the trajectory is the thing to assert.
Unexpected stop reasons are exits, not edge cases: an agent that treats a truncated response as success will eventually act on half a tool call.

Sources

Tool use with Claude — Anthropic, 2026docs
Function calling — OpenAI, 2026docs
ReAct: Synergizing Reasoning and Acting in Language Models — Yao et al., 2022paper
Building effective agents — Anthropic, 2024blog

#What real APIs actually send

#The offline program

#Tools: structured results, readable errors

#Validation that does not swallow bugs

#The loop and its exits

#Tests for the offline loop

#The same loop against Claude

#What this still leaves out