85 - Tools: Function Calling, JSON Schemas and Errors the Agent Can Read

A tool is a function plus a contract the model can read: a name, a description and a JSON schema for the arguments. The model executes nothing — it produces an object that says "call read_stock with {"sku": "SKU-11"}", and it is up to your code to decide whether to trust it. The whole craft follows from this asymmetry: the schema is a prompt, the arguments are untrusted input, and a tool's error is not an exception to propagate but a piece of information to return.

Version used and what I ran. The code was tried with openai-agents 0.24.0 (installed with no version constraint in a temporary virtual environment on 2026-11-12), pydantic 2.14.3 and Python 3.14.4. Pin the version: on this SDK, names shift between minor releases. To run the tools through the real SDK without a model, I use the same "fake Ollama" as the previous lesson (a simulated HTTP transport, no network). The call_tool(tool, arguments) function that appears in the examples makes the fake model run ONE tool and returns the text the model would receive; it is not shown in full, but it is the same construction as in that lesson. Where a block was not run, the text says so.

The analogy: the form to fill in at the counter

Picture a counter with a sign: "To request a certificate, fill in form A: field 1 tax code (16 characters), field 2 type of certificate (birth / residence / marriage)". The sign is the description, the form is the schema, and the clerk who checks the form before accepting it is the validation. A vague sign ("ask for what you need") produces forms filled in at random; a form with a free field where a choice among three was needed produces data you cannot process. And the clerk is needed anyway, because there is always someone who writes the date in the name field.

The journey of an argument, in one picture

A tool argument goes through five stages. It is worth seeing all of them because the security check is not where people think it is: the schema is at stage 2, but the validation that matters is at stage 5, inside your function.

The mechanism: from type hint to schema

When you decorate a function, the SDK inspects the signature and generates a JSON schema. I look at it on a tool with a Literal, a default and an integer, printing what the SDK exposes: name, description and parameter schema.

from lib import *

@function_tool
def search_orders(
    customer: str,
    status: Literal["open", "shipped", "cancelled"] = "open",
    max_results: int = 10,
) -> dict:
    """Search a customer's orders.

    Use this tool when the user asks about the status of one or more orders.
    Do not use it for returns: search_returns exists for those.
    """
    return {"customer": customer, "status": status, "max": max_results}

for t in [search_orders]:
    print("name       :", t.name)
    print("description:", t.description)
    print("parameters :", json.dumps(t.params_json_schema, indent=2, ensure_ascii=False))

# name       : search_orders
# description: Search a customer's orders.
#
# Use this tool when the user asks about the status of one or more orders.
# Do not use it for returns: search_returns exists for those.
# parameters : {
#   "properties": {
#     "customer": {
#       "title": "Customer",
#       "type": "string"
#     },
#     "status": {
#       "default": "open",
#       "enum": [
#         "open",
#         "shipped",
#         "cancelled"
#       ],
#       "title": "Status",
#       "type": "string"
#     },
#     "max_results": {
#       "default": 10,
#       "title": "Max Results",
#       "type": "integer"
#     }
#   },
#   "required": [
#     "customer",
#     "status",
#     "max_results"
#   ],
#   "title": "search_orders_args",
#   "type": "object",
#   "additionalProperties": false
# }

There are three lessons to read in this piece of JSON, and one of them contradicts a widespread idea. First: Literal becomes enum, and it is the most effective tool you have, because it closes the space of possible answers: an argument status: str will reach you as "in progress", "OPEN", "open?", while a Literal will not. Second: the description comes from the docstring and keeps the when not to use it; the line "do not use it for returns" is worth more than a paragraph of instructions in the system prompt, because it sits where the model looks while it decides. Third, the surprise: all three parameters appear in required, even status and max_results, which have a default. With the SDK's default settings (the "strict" schema, the same one we saw in the previous lesson with "strict": true) the defaults stay in the schema but the fields become mandatory. So it is not true that "defaults make the field optional" for the model: they make the value to use obvious, but the model still has to emit it. If you need a truly optional field, check your version's documentation on how to declare it.

Why the docstring is production code

Something worth saying explicitly, because it contradicts a deeply rooted habit: in ordinary programming the docstring is for humans, and if it is vague or missing the program works anyway. Not here. The docstring ends up in the prompt and is read by the model at the moment it decides whether to call you: together with the name and the schema, it is the only information that decision rests on.

The practical consequence is that a docstring has to be written by different criteria than usual. It does not describe the implementation, it describes when to use the tool and when not to. Compare these two versions of the same function.

# "Normal" docstring: correct for a human, useless for the model
@function_tool
def search_returns(customer: str, since: str) -> dict:
    """Queries the returns table filtering by customer and date, joining on orders."""
    ...

# Docstring as a prompt: says when, when not, and what to expect
@function_tool
def search_returns(customer: str, since: str) -> dict:
    """List a customer's RETURN requests starting from a date.

    Use this tool when the user talks about goods to send back, returns,
    replacements or refused parcels.
    Do NOT use it for refunds on delivered orders: search_orders exists for those.
    The customer name must be exact: if you do not know it, call
    find_customer first. The date must be in the format 2026-10-09.
    """
    ...

The second version contains four pieces of information that the first does not give and that the model uses: the user's intent that triggers it, the boundary with a neighbouring tool, the dependency on another tool, and the format of an argument. None of them is documentation of the implementation; all four reduce the number of wrong turns.

There are also two things that do not belong in the docstring, and they mirror the mistakes seen most often. Internal details do not belong ("it does a join on orders"): they are context paid for at every turn for information that changes none of the model's decisions. And threats do not belong ("do NOT call this tool more than once, it is important!"): the model has no counting mechanism you can rely on, and if a repeated call is a problem the defence is idempotency in the code, not emphasis in the text.

A small exercise that clarifies the whole point, and it is the print block I showed above: take your catalogue, print for each tool only the name, description and schema (params_json_schema) — that is, exactly what the model sees — and read them as if they were the help page of a program you do not know. In this version of the SDK, decorated tools expose name, description and params_json_schema; these are the names I verified and they may move.

The first pass of this reading always finds something: a description left empty, two tools that look like the same thing, an argument called val, an enum that had stayed a free string. It takes ten minutes, and they are the ten minutes with the best ratio of effort to improvement in the whole work on an agent.

Complex arguments: use a Pydantic model

For non-trivial input, declare a type instead of five loose parameters. You gain constraints expressed in the schema and an already validated object inside the function. But there is something I discovered by running the block, and it changes how you use it: look at what the model receives when the arguments violate a constraint.

from pydantic import BaseModel, Field
class ReorderRequest(BaseModel):
    sku: str = Field(pattern=r"^SKU-\d{2,6}$", description="Product code, e.g. SKU-11")
    quantity: int = Field(ge=1, le=1000, description="Pieces to order")
    urgency: Literal["standard", "urgent"] = "standard"

@function_tool
def create_reorder(request: ReorderRequest) -> dict:
    """Create a reorder proposal. Sends nothing to the supplier."""
    return {"status": "proposal_created", "sku": request.sku,
            "quantity": request.quantity, "urgency": request.urgency}

print(json.dumps(create_reorder.params_json_schema["$defs"]["ReorderRequest"]["properties"], ensure_ascii=False))

async def main():
    print(await call_tool(create_reorder, {"request": {"sku": "SKU-11", "quantity": 5}}))
    print(await call_tool(create_reorder, {"request": {"sku": "SKU-11", "quantity": 0}}))
    print(await call_tool(create_reorder, {"request": {"sku": "sku-11", "quantity": 5}}))
asyncio.run(main())

# {"sku": {"description": "Product code, e.g. SKU-11", "pattern": "^SKU-\\d{2,6}$", "title": "Sku", "type": "string"}, "quantity": {"description": "Pieces to order", "maximum": 1000, "minimum": 1, "title": "Quantity", "type": "integer"}, "urgency": {"default": "standard", "enum": ["standard", "urgent"], "title": "Urgency", "type": "string"}}
# {'status': 'proposal_created', 'sku': 'SKU-11', 'quantity': 5, 'urgency': 'standard'}
# An error occurred while running the tool. Please try again.
# An error occurred while running the tool. Please try again.

The three outputs say three things. The schema really does contain the constraints (pattern, minimum, maximum) and the field descriptions: the model reads them and tends to respect them. The valid call goes through. The two invalid ones produce no useful message for the model: with this version's default behaviour they receive only "An error occurred while running the tool. Please try again.", without saying which constraint was violated. The model can therefore retry in exactly the same way. When I replaced the error handling (next section) I obtained "Invalid JSON input for tool create_reorder", still without the detail of the field. In short: Pydantic constraints are a first barrier and a guide for the model, but they do not communicate the error; if you want the model to correct itself, the checks that matter have to be made inside the function, returning an error dictionary that names the field.

Errors: return them, do not raise them

This is the part that separates an agent that recovers from one that dies. A return value that describes the problem enters the history and the model reads it on the next turn. find_customer does this: if the name is not there, it returns suggestions and an instruction on how to fix it. I wrote it as a pure function (_impl) with a thin wrapper, so it can be tested without the SDK.

CUSTOMERS = {"ACME": 1, "Brambilla SRL": 2}
def find_customer_impl(name: str) -> dict:
    if name in CUSTOMERS:
        return {"ok": True, "id": CUSTOMERS[name]}
    similar = [c for c in CUSTOMERS if name.lower() in c.lower() or c.lower() in name.lower()]
    return {"ok": False, "error": "customer_not_found",
            "suggestions": similar or sorted(CUSTOMERS)[:5],
            "how_to_fix": "Call the tool again with one of the suggested names."}

print(find_customer_impl("Brambilla"))

# {'ok': False, 'error': 'customer_not_found', 'suggestions': ['Brambilla SRL'], 'how_to_fix': 'Call the tool again with one of the suggested names.'}

The how_to_fix field is not decorative: it is an instruction to the model, delivered at the exact moment it is needed. With an error like this, an agent that searched for "Brambilla" finds the suggestion "Brambilla SRL" and corrects itself in one turn (the model's correction is not tested here: the function is).

One boundary remains to be kept: not all errors are recoverable. An error on the data is returned (the model can fix it); an error in the system is raised, because the model can do nothing about it and retrying costs money. But the SDK, by default, does not propagate exceptions: it turns them into a generic message for the model, and the run continues. To make the run really stop on an infrastructure error you have to decide the rule yourself, with the failure_error_function parameter of function_tool. The following block defines one that converts data errors into text and re-raises infrastructure ones.

from lib import *
from agents import function_tool

class InfrastructureError(Exception):
    """Not recoverable by the model: the agent must stop."""

class ReorderRequest(BaseModel):
    sku: str = Field(pattern=r"^SKU-\d{2,6}$", description="Product code, e.g. SKU-11")
    quantity: int = Field(ge=1, le=1000)

def readable_error(context, error: Exception) -> str:
    """Turn the exception into a message the model can use to correct itself."""
    if isinstance(error, InfrastructureError):
        raise error                          # not recoverable: the run stops
    cause = error.__cause__ or error         # for invalid arguments, the detail is here
    lines = [e["msg"] + " (" + ".".join(str(p) for p in e["loc"]) + ")"
             for e in cause.errors()] if isinstance(cause, ValidationError) else [str(cause)]
    return "Invalid arguments or execution: " + "; ".join(lines)[:300]

@function_tool(failure_error_function=readable_error)
def create_reorder(request: ReorderRequest) -> dict:
    """Create a reorder proposal. Sends nothing to the supplier."""
    return {"status": "proposal_created", "quantity": request.quantity}

@function_tool(failure_error_function=readable_error)
async def read_invoice(number: str) -> dict:
    """Read an invoice from the accounting system."""
    if number == "down":
        raise InfrastructureError("accounting system unreachable")
    return {"ok": True}

@function_tool(failure_error_function=None)
def fragile(number: str) -> dict:
    """Always raises."""
    raise ValueError("boom")

async def main():
    for t, a in ((read_invoice, {"number": "down"}), (fragile, {"number": "x"})):
        try:
            print(await call_tool(t, a))
        except Exception as e:
            print(f"{t.name}: the run died with {type(e).__name__}: {e}")
asyncio.run(main())

# read_invoice: the run died with UserError: Error running tool read_invoice: accounting system unreachable
# fragile: the run died with UserError: Error running tool fragile: boom

Two lines, two ways to stop the run. With the custom function that re-raises, the InfrastructureError reaches your code as a UserError and the run stops: it is the intended behaviour for a database that is down. With failure_error_function=None, any exception does the same. The model, instead, never sees the exception. The opposite risk is using the conversion to a message for all errors, which is the way never to notice that the database is down: the model retries, the run continues, and the failure disappears from your logs. The generic message of the default behaviour is even worse, because it does not even say what to retry.

The tool is a system boundary: where security really lives

The five-stage diagram says the real check is at stage 5, inside the function. It is worth insisting, because it is the thing you understand late and it costs dearly: a tool's arguments are untrusted input, in the same sense that a web form field is. Not because the model is malicious, but because the arguments it generates depend on the text it has read — and that text can come from a user, from a web page, from a customer's email, from the result of another tool.

The consequence is that every classic application vulnerability shows up again, identical, with one extra step. Three concrete examples, because they are the ones you actually see.

The tool that runs SQL. A tool run_query(sql: str) is very convenient and is a SQL injection with extra steps: you are giving a text generator the ability to write arbitrary commands on your database. It is not solved with a prompt ("run only SELECT") because the prompt is not a control. It is solved by changing the shape of the tool: instead of accepting SQL, it accepts parameters and you build the query yourself.

To show the correct shape I ran the tool with a fake database that records the query and the parameters (in place of a driver such as asyncpg, whose $1 syntax I use here; I did not call a real database). The session context is an object of yours, passed to Runner.run(..., context=...) and invisible to the model.

from dataclasses import dataclass
from lib import *

@dataclass
class Session:
    run_id: str
    customer_id: int

class OrderFilter(BaseModel):
    customer_id: int = Field(ge=1)
    status: Literal["open", "shipped", "cancelled"] | None = None
    since: str | None = Field(default=None, pattern=r"^\d{4}-\d{2}-\d{2}$")
    limit: int = Field(default=20, ge=1, le=100)

class FakeDb:
    """In place of asyncpg: records the query and the parameters, returns two rows."""
    calls = []
    async def fetch(self, query, *parameters):
        self.calls.append((query, parameters))
        return [{"number": "A-1", "status": "open"}, {"number": "A-2", "status": "open"}]
db = FakeDb()

@function_tool
async def search_orders(context: RunContextWrapper[Session], filter: OrderFilter) -> dict:
    """Search a customer's orders. Returns at most 100 rows."""
    if filter.customer_id != context.context.customer_id:
        return {"ok": False, "error": "not_authorized",
                "how_to_fix": "You can only look at the current customer's orders."}
    rows = await db.fetch(
        "SELECT number, status FROM orders WHERE customer_id = $1 "
        "AND ($2::text IS NULL OR status = $2) ORDER BY date DESC LIMIT $3",
        filter.customer_id, filter.status, filter.limit)
    return {"ok": True, "orders": rows, "count": len(rows)}

async def main():
    ctx = Session(run_id="run-1", customer_id=7)
    print(await call_tool(search_orders, {"filter": {"customer_id": 7}}, ctx))
    print(await call_tool(search_orders, {"filter": {"customer_id": 8}}, ctx))
    print(db.calls[0])
asyncio.run(main())

# {'ok': True, 'orders': [{'number': 'A-1', 'status': 'open'}, {'number': 'A-2', 'status': 'open'}], 'count': 2}
# {'ok': False, 'error': 'not_authorized', 'how_to_fix': "You can only look at the current customer's orders."}
# ('SELECT number, status FROM orders WHERE customer_id = $1 AND ($2::text IS NULL OR status = $2) ORDER BY date DESC LIMIT $3', (7, None, 20))

The detail that matters more than any other is the line on customer_id. The model can put any number in it, because it is a field of the schema: the only defence is to compare it with the identity that comes from the authenticated session, which lives in the run context and is not visible to the model. The output shows the two paths: customer 7, the session's one, gets the rows; customer 8 receives "not_authorized". The third line shows the query with the parameters separated from the SQL text: no string from the model is concatenated into the query. This is the general form: the fields that determine authorization must not be in the schema, and if for convenience they are, they must be checked against a source the model does not control.

The tool that reads files. A read_file(path: str) without normalization is a path traversal: a ../../etc/passwd or an absolute path is enough. The defence is to resolve the path and verify that it stays inside the permitted root, after resolving symbolic links. I tried it in a temporary folder with a legitimate file, a CSV, and a symbolic link pointing outside the root.

import os, tempfile
from pathlib import Path
from lib import *

folder = tempfile.mkdtemp()
ROOT = Path(folder).resolve()
(ROOT / "note.md").write_text("hello", encoding="utf-8")
(ROOT / "data.csv").write_text("a,b", encoding="utf-8")
os.symlink("/etc/hostname", ROOT / "outside.md")           # link pointing outside

@function_tool
def read_document(name: str) -> dict:
    """Read a document from the archive. Only .md or .txt files of the archive."""
    candidate = (ROOT / name).resolve()                   # resolves .. and symlinks
    if not candidate.is_relative_to(ROOT):                # Python 3.9+
        return {"ok": False, "error": "path_outside_archive"}
    if candidate.suffix not in {".md", ".txt"}:
        return {"ok": False, "error": "extension_not_allowed"}
    if not candidate.is_file():
        return {"ok": False, "error": "not_found", "name": name}
    text = candidate.read_text(encoding="utf-8", errors="replace")[:20_000]
    return {"ok": True, "name": name, "text": text,
            "truncated": candidate.stat().st_size > 20_000}

async def main():
    for name in ("note.md", "../etc/passwd", "data.csv", "outside.md", "missing.md"):
        print(f"{name:14s}", await call_tool(read_document, {"name": name}))
asyncio.run(main())

# note.md        {'ok': True, 'name': 'note.md', 'text': 'hello', 'truncated': False}
# ../etc/passwd  {'ok': False, 'error': 'path_outside_archive'}
# data.csv       {'ok': False, 'error': 'extension_not_allowed'}
# outside.md     {'ok': False, 'error': 'path_outside_archive'}
# missing.md     {'ok': False, 'error': 'not_found', 'name': 'missing.md'}

Five cases, five outcomes. The legitimate file is read. The path with .. and the symbolic link to the outside both end up in "path_outside_archive", because resolve() follows the link before the is_relative_to check: had I checked the name before resolving, the link would have got through. The CSV is rejected by extension, and the nonexistent file gets a specific error instead of an exception.

The tool that makes an HTTP request. A download(url: str) is a Server-Side Request Forgery: a model can be induced to ask for http://169.254.169.254/ or an address on your internal network, and your server, which has access to things the user does not, carries it out. The defence is a list of allowed domains, not a list of forbidden addresses, because deny lists are bypassed by a DNS that resolves wherever whoever controls it wants. I did not write or run an example of this tool.

The common thread of the three defences is a principle worth stating because it guides every later decision: least privilege applied to the shape of the tool. Not "give the tool minimal permissions" (which is right but generic), but: design the tool so that the dangerous thing cannot be expressed. A tool that accepts SQL can do anything; a tool that accepts a typed filter cannot. The difference is not how much validation you write: it is what you put in the schema.

Designing the catalogue: the two-tool test

The granularity of tools is the design decision with the most effect on the number of turns and on the error rate, and it is taken with a test that takes thirty seconds. Take two tools from your catalogue, read only the name and description (not the code), and ask yourself: in this situation, would I know which one to pick? If you hesitate, the model will hesitate too, and its hesitation costs a turn or a wrong choice.

A few recurring cases, with the cure.

Near-twin tools. find_customer_by_name, find_customer_by_email, find_customer_by_vat_number: three tools that do the same thing with a different key. Each description takes up context at every turn, and the model has to make a choice that adds no information. They merge into one tool with a typed parameter: find_customer(value: str, kind: Literal["name","email","vat"]).

A tool that does two things. manage_order(action: Literal["read","cancel","edit"], ...) bundles a harmless read and two irreversible writes under the same name. It is convenient and wrong for two independent reasons: the model picks the action with the same lightness with which it fills in any field, and you cannot give different permissions to different operations. Reads and writes go in separate tools, always.

A tool that is too generic. run_operation(name: str, parameters: dict) is a way of not designing the catalogue: the schema says nothing, the model does not know what it can ask for, and all the validation ends up at run time. You see it in projects that want "one tool for everything", and it produces agents that fail in ways impossible to diagnose.

A practical rule that works: one tool per user intention, not per endpoint of your backend. If your back office exposes six calls to find out whether an order can be shipped, the tool exposes one, which internally makes six. The model does not need to know your architecture — and every detail of your architecture you let it know is context you pay for and surface on which it can go wrong.

The return value is context, and context is paid n times

This is the part of tool design that is ignored most often, and the only one where ten minutes of work produce a saving. The reason is the mechanism seen in the lesson on anatomy: what a tool returns enters the state, and the state is sent back to the model at every following turn. A tool that returns 3000 tokens on the first of four turns therefore appears in four calls: 3000 tokens generated and three re-sends, for 12,000 tokens attributable to that tool.

A worked example, with estimated numbers, not measured with a tokenizer: a tool that searches orders and returns the full database row, twenty columns of which the model uses four, over twenty orders.

"SELECT *" version:  20 orders x 20 columns ~ 1600 tokens (estimate)
  turn 1 the tool returns, turns 2-4 re-sent: 1600 x 4 = 6400

"useful fields" version: 20 orders x 4 columns ~ 320 tokens (estimate, 1/5)
  same calculation: 320 x 4 = 1280

The ratio is that of the columns, five to one, and holds regardless of the absolute values, which depend on the content and on the tokenizer. The saving is not only economic: less irrelevant material in the context means fewer chances to confuse the model.

From here, three rules, with the technical reason beside each.

Return the fields that are needed, never the whole row. The criterion is not "what might be useful" but "what is needed for the decision this agent has to make". If the agent decides whether to reorder, it does not need the name of the warehouse clerk who did the last inventory.

Paginate, and declare that you are paginating. This is the rule that avoids a category of lies told in perfect good faith. If the tool returns ten results out of four hundred and twenty-eight and does not say so, the model concludes there are ten and reports it to the user with confidence. The field that solves it is trivial:

return {"orders": rows[:10],
        "shown": len(rows[:10]),
        "total_available": total,
        "note": ("Showing the first 10 of "
                 f"{total}. If you need a precise subset, refine the filters.")}

That natural-language note is not redundant with the numbers: it is the instruction on what to do, delivered at the exact point where it is needed, and in practice it is what makes the difference between a model that refines the search and a model that settles.

Descriptive keys and explicit units. {"weight": 3.2} is ambiguous — kilos, grams, pounds? — and the model will resolve the ambiguity by guessing. {"weight_kg": 3.2} leaves no room. It holds for currencies (amount_eur), for dates (preferably in ISO 8601, the only format that cannot be misread between day and month) and for booleans, where {"shippable": false, "not_shippable_reason": "incomplete address"} tells the model what to do and {"status": 0} does not.

One last measure that costs nothing and saves time: always return the same schema, even on error. A tool that returns a list on success and a string on error forces the model to handle two different shapes. A tool that always returns {"ok": bool, ...} gives it a single place to look. It is the same logic as the response envelope used for APIs towards the outside, and it holds for the same reasons.

Tools that write: idempotency and confirmation

A tool that reads can be called ten times with no damage. A tool that writes — sends an order, sends an email, deletes a record — called twice does the damage twice. And the model will repeat calls: because the result is not clear to it, because a timeout triggered a retry, because the history was truncated and it does not see that it has already acted.

The defence is not a prompt ("do not call this tool twice"): it is idempotency. An operation is idempotent when running it N times has the same effect as running it once. The standard way to get it is a deduplication key decided by the caller.

The following block defines send_order with a deduplication key and calls it twice with the same data through the SDK, with a session that carries the run identifier.

import hashlib
from dataclasses import dataclass
from lib import *

@dataclass
class Session:
    run_id: str
    customer_id: int

ORDERS_SENT: dict[str, dict] = {}

class Order(BaseModel):
    sku: str = Field(pattern=r"^SKU-\d{2,6}$")
    quantity: int = Field(ge=1, le=1000)
    supplier: str

def operation_key(run_id: str, o: Order) -> str:
    raw = f"{run_id}|{o.sku}|{o.quantity}|{o.supplier}"
    return hashlib.sha256(raw.encode()).hexdigest()[:16]

@function_tool
def send_order(context: RunContextWrapper[Session], order: Order) -> dict:
    """Send an order to the supplier. WARNING: not reversible.

    Calling this tool twice with the same data does NOT create two orders.
    """
    key = operation_key(context.context.run_id, order)
    if key in ORDERS_SENT:                             # already done: identical outcome
        return {"ok": True, "duplicate": True, **ORDERS_SENT[key]}
    outcome = {"order_number": f"ORD-{key}", "sku": order.sku, "quantity": order.quantity}
    ORDERS_SENT[key] = outcome
    return {"ok": True, "duplicate": False, **outcome}

async def main():
    ctx = Session(run_id="run-1", customer_id=7)
    o = {"order": {"sku": "SKU-11", "quantity": 5, "supplier": "ACME"}}
    print(await call_tool(send_order, o, ctx))
    print(await call_tool(send_order, o, ctx))
    print(len(ORDERS_SENT), "order(s) recorded")
asyncio.run(main())

# {'ok': True, 'duplicate': False, 'order_number': 'ORD-433e96f6500c31a1', 'sku': 'SKU-11', 'quantity': 5}
# {'ok': True, 'duplicate': True, 'order_number': 'ORD-433e96f6500c31a1', 'sku': 'SKU-11', 'quantity': 5}
# 1 order(s) recorded

Two calls, a single order recorded: the second returns the same order_number with duplicate: True. Two details. The first: RunContextWrapper is the way the SDK hands you an object of yours (here with the run identifier) without exposing it to the model: it does not appear in the schema, so the model can neither invent nor tamper with it; it is the right place for the authenticated user, the session identifier, permissions. The second: returning duplicate: True instead of an error is deliberate, because the model reads "ok", does not retry, and you have not sent two orders. The in-memory dictionary is fine for the example; in production you need a table with a unique constraint.

For truly expensive operations idempotency is not enough: you need a human confirmation. The clean way is to split into two tools, one that prepares and one that executes; _really_send here is a placeholder that returns a dictionary and sends nothing.

# (same definitions of Order and operation_key as in the previous block)
DRAFTS: dict[str, Order] = {}

def _really_send(order: Order, approved_by: str) -> dict:
    return {"ok": True, "sent": order.sku, "approved_by": approved_by}

@function_tool
def prepare_order(order: Order) -> dict:
    """Prepare an order and return a confirmation token. Sends NOTHING."""
    token = operation_key("draft", order)
    DRAFTS[token] = order
    return {"token": token, "summary": f"{order.quantity} x {order.sku}",
            "note": "Show the summary to the user and ask for confirmation. "
                    "Call confirm_order ONLY if the user explicitly approves."}

@function_tool
def confirm_order(token: str, approved_by: str) -> dict:
    """Send a prepared order. Requires the token from prepare_order."""
    order = DRAFTS.pop(token, None)
    if order is None:
        return {"ok": False, "error": "token_expired_or_missing",
                "how_to_fix": "Call prepare_order again."}
    return _really_send(order, approved_by)

async def main():
    o = {"order": {"sku": "SKU-11", "quantity": 5, "supplier": "ACME"}}
    await call_tool(prepare_order, o)
    token = list(DRAFTS)[0]
    print(await call_tool(confirm_order, {"token": "made_up", "approved_by": "mario"}))
    print(await call_tool(confirm_order, {"token": token, "approved_by": "mario"}))
asyncio.run(main())

# {'ok': False, 'error': 'token_expired_or_missing', 'how_to_fix': 'Call prepare_order again.'}
# {'ok': True, 'sent': 'SKU-11', 'approved_by': 'mario'}

A token invented by the model is rejected; the one produced by prepare_order works, and only once, because pop consumes it. The structural point: the second tool requires a token that only the first can produce. The model cannot skip the step, because it does not know the token. It is a protocol constraint, not an invitation in the prompt, and the difference between the two is the whole difference between a safe system and a system that trusts. One limit remains: the token prevents skipping, it does not guarantee that the user really approved; that guarantee lives in the interface that collects the confirmation, not in the model.

Testing a tool (it is ordinary code, and that is the point)

Tools are the deterministic part of an agent: pure or nearly pure functions, with defined inputs and outputs, tested without a model and without spending anything. Keep the logic in an ordinary function (read_stock_impl) and apply @function_tool to a thin wrapper, so the tests do not depend on how the SDK wraps functions, which is one of the things that changes between versions. The block uses pytest in the usual form; I ran the same assertions as plain functions, without pytest, and report that output.

import asyncio, json
from lib import *

CUSTOMERS = {"ACME": 1, "Brambilla SRL": 2}
WAREHOUSE = {"SKU-11": {"stock": 4, "threshold": 10}}

def read_stock_impl(sku: str) -> dict:
    row = WAREHOUSE.get(sku)
    if row is None:
        return {"error": "unknown_sku", "sku": sku, "known": sorted(WAREHOUSE)}
    return {"sku": sku, **row}

@function_tool
def read_stock(sku: str) -> dict:
    """Current stock and reorder threshold of a SKU."""
    return read_stock_impl(sku)               # thin wrapper: the logic lives in the _impl

def find_customer_impl(name: str) -> dict:
    if name in CUSTOMERS:
        return {"ok": True, "id": CUSTOMERS[name]}
    similar = [c for c in CUSTOMERS if name.lower() in c.lower() or c.lower() in name.lower()]
    return {"ok": False, "error": "customer_not_found",
            "suggestions": similar or sorted(CUSTOMERS)[:5],
            "how_to_fix": "Call the tool again with one of the suggested names."}

class Order(BaseModel):
    sku: str = Field(pattern=r"^SKU-\d{2,6}$")
    quantity: int = Field(ge=1, le=1000)
    supplier: str

def test_unknown_sku_returns_error_does_not_raise():
    outcome = read_stock_impl("SKU-DOES-NOT-EXIST")
    assert outcome["error"] == "unknown_sku"
    assert "known" in outcome                  # the suggestion is there

def test_partial_customer_has_suggestion():
    assert find_customer_impl("Brambilla")["suggestions"] == ["Brambilla SRL"]

def test_schema_rejects_malformed_skus():
    for bad in ["", "11", "sku-11", "SKU-", "SKU-1234567"]:
        try:
            Order(sku=bad, quantity=1, supplier="ACME")
        except ValidationError:
            continue
        raise AssertionError(f"accepted: {bad!r}")

for t in (test_unknown_sku_returns_error_does_not_raise,
          test_partial_customer_has_suggestion, test_schema_rejects_malformed_skus):
    t(); print("ok:", t.__name__)

# ok: test_unknown_sku_returns_error_does_not_raise
# ok: test_partial_customer_has_suggestion
# ok: test_schema_rejects_malformed_skus

Note the first assertion, which is a design choice made executable: the test verifies that the tool does not raise and that the error contains the suggestion. If someone tomorrow "fixes" that function by turning the dictionary into an exception, the test goes red and explains why. The third test checks the Pydantic model's constraints, that is, the barrier discussed above: it rejects the five wrong forms of SKU, even though, as you have seen, the model would not receive the reason.

Trade-offs

Four choices pull in opposite directions. Few generic tools versus many specific tools: few tools reduce choice errors and the context taken up, but each tool becomes a function with many branches to validate; many specific tools are simple to write and hard to choose well. Tight schemas versus permissive schemas: tight ones (enum, pattern, range) mean more rejections, more turns and less damage; permissive ones, less friction and more junk entering the system; for tools that write, always tighten.

Long descriptions versus context: each tool's description is sent at every turn, so ten tools with five lines of docstring are a recurring fixed cost; the line that prevents a wrong call is worth it, the one that repeats the obvious is not. Synchronous versus asynchronous tools: synchronous ones are simpler but block the event loop, asynchronous ones are correct under load and require async clients throughout the chain.

The traps you really fall into

1. Treating arguments as trusted data. They are generated by a model from text that may come from a user or a web page. A tool run_sql(query: str) is a SQL injection with extra steps; a tool read_file(path: str) without normalization is a path traversal. The schema is not a security check: you write the check inside the function.
2. Non-idempotent tools without a deduplication key. The model can repeat a call (because the result is not clear to it, because there was a retry). If the tool sends an order, the order goes out twice. Solution: accept an operation identifier and discard duplicates.
3. Overlapping names. search_order and find_order in the same catalogue guarantee random choices. One intention, one tool, one name.
4. Putting secrets in the schema. A parameter api_key: str will be made up by the model. Credentials are read from the environment inside the function.
5. Returning free text instead of data. "I found three orders, the most recent is from Tuesday" forces the model to re-parse prose. Return structures; it writes the prose itself at the end.
6. Forgetting that the tool's return is untrusted text. If a tool reads a web page or a customer's email, that content enters the context and may contain instructions. It is the subject of the lesson on prompt injection, and it is solved at the level of architecture, not of schema.

The six traps have two roots. The first four come from forgetting that the model is an author of input: untrusted arguments, repeated calls, ambiguous names, secrets it makes up. The last two come from forgetting that the tool is also a producer of context: free text that has to be re-parsed, external content that can carry instructions. In both cases the defence lies in the shape of the tool and in the code around it, not in the prompt.

What you can do now

You can read the schema the SDK generates from type hints and the docstring, you know that defaults stay mandatory in the strict schema and that Pydantic validation errors do not reach the model with the detail, you can write tools that return useful errors and tools that stop on system errors, and you can close the surface of a dangerous tool (query, file, HTTP request) and make it idempotent. In the next lesson, Structured Output, we turn the arrow around: no longer data going into the tool, but typed data coming out of the agent.