An agent that uses tools

import anthropic

client = anthropic.Anthropic()
MODEL = "claude-opus-5"

ORDERS = {
    "ada@example.com": [
        {"id": "A-1001", "item": "keyboard", "status": "delivered", "total": 89.0},
        {"id": "A-1042", "item": "monitor arm", "status": "in transit", "total": 54.0},
    ],
}

SEARCH_ORDERS = {
    "name": "search_orders",
    # The description is prompt, not documentation. It is the only thing the
    # model reads when deciding whether this tool is the right one.
    "description": (
        "Look up the orders belonging to one customer, by email address. "
        "Returns at most 20 orders, newest first. Use this before answering "
        "any question about what a customer bought or where an order is."
    ),
    "input_schema": {
        "type": "object",
        "properties": {
            "email": {"type": "string", "description": "The customer's email address."},
        },
        "required": ["email"],
        "additionalProperties": False,
    },
    # With strict, the arguments are guaranteed to match the schema, so you
    # are validating your own logic rather than the model's JSON.
    "strict": True,
}

def search_orders(email: str) -> list[dict]:
    return ORDERS.get(email.lower(), [])[:20]

response = client.messages.create(
    model=MODEL,
    max_tokens=4096,
    tools=[SEARCH_ORDERS],
    messages=[{"role": "user", "content": "Where is ada@example.com's monitor arm?"}],
)

for block in response.content:
    if block.type == "tool_use":
        print(block.name, block.input)   # search_orders {'email': 'ada@example.com'}

import json

MAX_STEPS = 12

class StepBudgetExceeded(RuntimeError):
    pass

def call_tool(block, dispatch):
    """Turn one tool_use block into one tool_result block."""
    fn = dispatch.get(block.name)
    if fn is None:
        # The model invented a tool. Say so in a result rather than crashing:
        # it can recover from being told, and cannot recover from a traceback.
        return {
            "type": "tool_result", "tool_use_id": block.id, "is_error": True,
            "content": f"No tool named {block.name}. Available: {sorted(dispatch)}",
        }
    output = fn(**block.input)
    return {
        "type": "tool_result",
        "tool_use_id": block.id,          # must match, or the API rejects the turn
        "content": json.dumps(output)[:4000],
    }

def run(task: str, tools, dispatch, max_steps: int = MAX_STEPS):
    messages = [{"role": "user", "content": task}]

    for step in range(max_steps):
        response = client.messages.create(
            model=MODEL, max_tokens=4096, tools=tools, messages=messages,
        )
        # Append the whole content list, not just the text: the tool_use blocks
        # have to come back or the results below refer to nothing.
        messages.append({"role": "assistant", "content": response.content})

        if response.stop_reason != "tool_use":
            return response, messages     # the only clean exit

        results = [
            call_tool(b, dispatch) for b in response.content if b.type == "tool_use"
        ]
        # All results from one turn go back in a single user message.
        messages.append({"role": "user", "content": results})

    raise StepBudgetExceeded(f"no answer after {max_steps} steps")

from concurrent.futures import ThreadPoolExecutor

TRACK_SHIPMENT = {
    "name": "track_shipment",
    "description": (
        "Get the current location and expected delivery date for one shipment, "
        "by tracking number. Does NOT accept an order id or an email address — "
        "use search_orders first to find the tracking number."
    ),
    "input_schema": {
        "type": "object",
        "properties": {"tracking": {"type": "string"}},
        "required": ["tracking"],
        "additionalProperties": False,
    },
    "strict": True,
}

def dispatch_all(blocks, dispatch, workers=4):
    """Tool calls in one turn are independent, so run them at once.
    Order is preserved: results must line up with the blocks that asked."""
    with ThreadPoolExecutor(max_workers=workers) as pool:
        return list(pool.map(lambda b: call_tool(b, dispatch), blocks))

import random, time
import httpx

class ToolFailed(Exception):
    """A failure the model should be told about."""

def with_retries(fn, *, attempts=3, timeout=10.0):
    for attempt in range(attempts):
        try:
            return fn(timeout=timeout)
        except (httpx.TimeoutException, httpx.TransportError) as exc:
            if attempt == attempts - 1:
                raise ToolFailed(f"{type(exc).__name__} after {attempts} attempts") from exc
            # Exponential backoff with jitter, so concurrent agents do not
            # retry in lockstep and re-create the outage they are waiting out.
            time.sleep(2**attempt + random.random())

def call_tool(block, dispatch):
    fn = dispatch.get(block.name)
    if fn is None:
        return _error(block, f"No tool named {block.name}. Available: {sorted(dispatch)}")
    try:
        output = fn(**block.input)
    except ToolFailed as exc:
        return _error(block, f"{block.name} is unavailable: {exc}. "
                             "Try a different approach or tell the user it cannot be checked.")
    except (TypeError, ValueError) as exc:
        # Bad arguments. Recoverable, and only the model can fix it.
        return _error(block, f"Invalid arguments for {block.name}: {exc}")
    return {"type": "tool_result", "tool_use_id": block.id,
            "content": json.dumps(output)[:4000]}

def _error(block, message: str):
    return {"type": "tool_result", "tool_use_id": block.id,
            "is_error": True, "content": message}

count = client.messages.count_tokens(model=MODEL, tools=tools, messages=messages)
print(step, count.input_tokens)

# Or read it from the response you already paid for:
print(response.usage.input_tokens, response.usage.output_tokens)

def trim(messages, keep_last=3):
    """Stub out the content of old tool results, keeping their ids.

    The first user message is the task and is never dropped. Recent results
    are what the model is working from. Everything older becomes a stub,
    which costs a few tokens instead of a few thousand."""
    cutoff = len(messages) - keep_last * 2      # one assistant + one user turn per step
    trimmed = []
    for i, msg in enumerate(messages):
        content = msg["content"]
        if i == 0 or i >= cutoff or not isinstance(content, list):
            trimmed.append(msg)
            continue
        trimmed.append({
            "role": msg["role"],
            "content": [
                {**b, "content": "[earlier result, trimmed]"}
                if isinstance(b, dict) and b.get("type") == "tool_result" else b
                for b in content
            ],
        })
    return trimmed

import json, time, uuid
from pathlib import Path

TRACE = Path("traces.jsonl")

def trace(run_id, step, kind, **fields):
    record = {"run_id": run_id, "step": step, "kind": kind, "ts": time.time(), **fields}
    with TRACE.open("a") as f:
        f.write(json.dumps(record, default=str) + "\n")

# In the loop, around the model call:
t0 = time.perf_counter()
response = client.messages.create(model=MODEL, max_tokens=4096,
                                  tools=tools, messages=messages)
trace(run_id, step, "model",
      stop_reason=response.stop_reason,
      input_tokens=response.usage.input_tokens,
      output_tokens=response.usage.output_tokens,
      ms=round((time.perf_counter() - t0) * 1000))

# And around each tool call:
t0 = time.perf_counter()
result = call_tool(block, dispatch)
trace(run_id, step, "tool",
      name=block.name,
      args=block.input,                       # the arguments, not just the name
      ok=not result.get("is_error"),
      bytes=len(result["content"]),
      ms=round((time.perf_counter() - t0) * 1000))

IRREVERSIBLE = {"issue_refund", "send_email", "delete_record"}

def call_tool(block, dispatch, approve):
    if block.name in IRREVERSIBLE:
        allowed, reason = approve(block.name, block.input)
        if not allowed:
            return _error(
                block,
                f"A human declined this action: {reason}. Do not attempt it again. "
                "Explain to the user that it needs approval.",
            )
    ...

def approve_at_terminal(name, args):
    print(f"\n{name}({json.dumps(args, indent=2)})")
    answer = input("approve? [y/N] ").strip().lower()
    return (answer == "y", "declined at the terminal")