Tool Use and Function Calling

import anthropic

client = anthropic.Anthropic()          # reads the API key from the environment

tools = [{
    "name": "get_order",
    "description": "Look up one order by its id. Use this before making any "
                   "statement about an order's contents, status or lateness.",
    "input_schema": {
        "type": "object",
        "properties": {"order_id": {"type": "integer", "description": "The order number"}},
        "required": ["order_id"],
        "additionalProperties": False,
    },
}]

messages = [{"role": "user", "content": "Was order 88213 late?"}]
reply = client.messages.create(
    model="claude-opus-5-5", max_tokens=16000, tools=tools, messages=messages,
)

# reply.stop_reason == "tool_use"
# reply.content contains: {"type": "tool_use", "id": "toolu_01", "name": "get_order",
#                          "input": {"order_id": 88213}}

call = next(block for block in reply.content if block.type == "tool_use")
result = get_order(**call.input)        # YOUR code runs the function

messages.append({"role": "assistant", "content": reply.content})
messages.append({"role": "user", "content": [
    {"type": "tool_result", "tool_use_id": call.id, "content": str(result)},
]})
# Call the model again with the longer messages list to get its answer.

refund_tool = {
    "name": "refund_delivery_fee",
    "description": (
        "Refund the delivery fee of ONE order to the customer's wallet. "
        "Use only after get_order shows the order was more than 30 minutes late. "
        "Do not use for item refunds; use refund_item for those."
    ),
    "strict": True,                      # inputs are guaranteed to match the schema
    "input_schema": {
        "type": "object",
        "properties": {
            "order_id": {"type": "integer", "description": "From get_order, e.g. 88213"},
            "reason": {
                "type": "string",
                "enum": ["late_delivery", "rider_no_show", "goodwill"],
                "description": "Why the fee is being refunded",
            },
        },
        "required": ["order_id", "reason"],
        "additionalProperties": False,
    },
}

import json

TOOLS = {"get_order": get_order, "refund_delivery_fee": refund_delivery_fee}

def execute(call, user) -> dict:
    """Run one tool request and return a tool_result block."""
    def result(text: str, error: bool = False) -> dict:
        return {"type": "tool_result", "tool_use_id": call.id,
                "content": text, "is_error": error}

    fn = TOOLS.get(call.name)
    if fn is None:                                   # the model invented a tool
        return result(f"No tool named {call.name}. Available: {sorted(TOOLS)}", True)

    problem = validate(call.name, call.input)        # check against the JSON schema
    if problem:
        return result(f"Invalid input: {problem}", True)

    if not allowed(user, call.name, call.input):     # permission, decided by code
        return result("Not permitted for this customer. Use hand_over.", True)

    try:
        value = fn(**call.input, user=user)          # identity comes from the session
    except TimeoutError:
        return result("The orders service timed out. You may try once more.", True)

    return result(json.dumps(value)[:4000])          # short, bounded observation

import time

def call_with_retry(fn, *, attempts: int = 3, base_delay: float = 0.5, **kwargs):
    """Retry passing faults only. Waits 0.5s, 1s, 2s between tries."""
    for attempt in range(attempts):
        try:
            return fn(**kwargs)
        except (TimeoutError, RateLimited):
            if attempt == attempts - 1:
                raise
            time.sleep(base_delay * 2 ** attempt)
        # ValidationError and PermissionDenied are NOT caught: retrying cannot fix them.

from concurrent.futures import ThreadPoolExecutor

READ_ONLY = {"get_order", "get_policy", "get_customer_history"}

def run_calls(calls, user) -> list[dict]:
    """Read tools run together; anything that writes runs one at a time."""
    if all(call.name in READ_ONLY for call in calls):
        with ThreadPoolExecutor() as pool:
            return list(pool.map(lambda c: execute(c, user), calls))
    return [execute(call, user) for call in calls]

# All results go back in ONE user message, one tool_result per request.

Q: What are tools?

Functions in your code that the model can ask to have run, each described by a name, a description and an input schema. They let the model read live data, compute exactly and take actions.

Q: How does function calling work?

You send tool definitions with the conversation. The model replies with a structured request: tool name, id, JSON inputs. Your code runs the function and sends a result with the same id. The model then continues. The model never executes anything.

Q: What makes a good tool?

Shaped around a task, not an API endpoint; one clear purpose; a description saying when to use it and when not; narrow typed inputs with enums; short results; error messages that say how to fix the call.

Q: How do you handle a tool error?

Return it to the model as a result marked as an error, with a message it can act on. Retry passing faults in code with backoff. Never retry permission failures. Make writes idempotent so a retry cannot repeat the effect.

Q: What if the model calls a tool that does not exist?

Do not crash. Return an error result naming the real tools. Strict schemas and a short tool list make it rare; it is counted as a metric and used to improve descriptions.

Q: When do you run tools in parallel?

When they are independent reads. It cuts latency to the slowest call and reduces turns. Writes, or calls that depend on each other, run in sequence.

Q: How do you support five hundred tools?

Do not show them all. Route by category or retrieve the few relevant tool definitions by search, keep a small always-on core, and measure selection accuracy.

Q: Where do permissions live?

In the tool and the executor, not the prompt. Identity comes from the session, limits are checked in code against real data, and risky actions wait for approval.

Your task. Write a small tool layer in Python with no agent framework: register tools, validate inputs, run them, and return well-shaped results and errors.

Steps

1. Write three plain functions over an in-memory dictionary of fake orders: get_order(order_id), calculate(expression) and refund_delivery_fee(order_id, reason).
2. Write a JSON Schema for each, with descriptions, required fields, an enum for reason and additionalProperties: false.
3. Write execute(name, inputs, user) that rejects unknown tools, validates inputs (the jsonschema package does this), checks the order belongs to user, and returns {ok, content}.
4. Make refund_delivery_fee refuse a second refund for the same order.
5. Write six tests that call execute directly: a good call, an unknown tool, a missing field, a wrong type, another user's order, and a repeated refund.

You are done when

• All six tests pass without any model being called.
• Every error message says what was wrong and what to do instead.
• No tool takes the user's identity from its inputs.

Stretch. Add run_calls that executes read-only calls in parallel and writes in sequence, and time three look-ups both ways.