08 - Advanced Tool Calling: APIs, Web and Custom Tools

The problem: the model proposes, your code disposes

When an agent “uses a tool”, the model executes nothing: it produces a message with the name of the tool and the arguments, and you are the one who runs it and sends the result back. In the format documented by Ollama the call is a tool_calls element with type: function, the name and the arguments; the answer comes back in a message with the role tool and the field tool_name. The MCP protocol calls it tools/call, and Anthropic's API uses tool_result blocks.

The labels change, the structure is the same, and that is why this article is about your code and not about a vendor: everything that sits between the model's proposal and the effect on the world (the schema, the validation, the permissions, the error handling) is yours. We saw the message format in Fundamentals: OODA Loop, ReAct and Tool Calling; here we go further, towards tools that touch APIs and databases.

The listings are run locally with Python 3.14.4. Five use only the standard library; the second uses jsonschema 4.26.0, installed in a virtual environment outside the repository. No model was called and no service was reached: the “arguments produced by the model” are hand-written dictionaries, and where something would need a real service I say so.

Defining the tool: the schema is the contract

A tool has a name, a description and a schema of its parameters in JSON Schema: it is the shape taken by all the definitions mentioned above (in the MCP protocol the field is called inputSchema). The model sees only this text, so the description is the only place where it can learn when to use the tool.

Anthropic's documentation is explicit: providing extremely detailed descriptions is, in its words, the most important factor in performance, and a good description explains what the tool does, when it should be used (and when not), what each parameter means and what the limits are, for example what it does not return.

Writing the schema by hand for fifty functions is tedious and drifts out of sync with the code. That is why the Ollama Python SDK, in its documentation, accepts a function directly and derives the definition from its docstring. The listing does the same thing in miniature: a decorator reads the signature, the types and the default values with inspect, builds the definition and registers it; parameters without a default value become required, and additionalProperties: false forbids invented parameters.

import inspect, json

TYPES = {str: "string", int: "integer", float: "number", bool: "boolean"}
REGISTRY = {}

def tool(function):
    """Builds the definition from the function: name, docstring, types and default values."""
    signature = inspect.signature(function)
    properties, required = {}, []
    for name, p in signature.parameters.items():
        properties[name] = {"type": TYPES[p.annotation]}
        if p.default is inspect.Parameter.empty:
            required.append(name)
        else:
            properties[name]["default"] = p.default
    definition = {"type": "function", "function": {
        "name": function.__name__,
        "description": inspect.getdoc(function),
        "parameters": {"type": "object", "properties": properties,
                       "required": required, "additionalProperties": False}}}
    REGISTRY[function.__name__] = (function, definition)
    return function

@tool
def search_orders(customer: str, status: str = "open", limit: int = 5):
    """Searches a customer's orders by status. Changes nothing: to cancel
    an order use cancel_order. Returns at most `limit` orders."""
    return []

definition = REGISTRY["search_orders"][1]["function"]
print(definition["name"])
print(definition["description"].replace("\n", " "))
print(json.dumps(definition["parameters"], ensure_ascii=False))

# Output:
# search_orders
# Searches a customer's orders by status. Changes nothing: to cancel an order use cancel_order. Returns at most `limit` orders.
# {"type": "object", "properties": {"customer": {"type": "string"}, "status": {"type": "string", "default": "open"}, "limit": {"type": "integer", "default": 5}}, "required": ["customer"], "additionalProperties": false}

The output shows what you get for free and what you do not. Name, description, types and required parameters come from the code; status and limit carry their default value. What is missing are the constraints that Python types do not express: nothing says that limit cannot be 5000, or that status is only “open” or “closed”.

Those constraints must be added to the schema (or to a second validation copy), and that is the work of the next section. Another consequence: the description is the docstring, so revising the docstring means changing the agent's behavior, and it must be treated as code, with review and tests (as we do in Testing and Evaluation).

An effective description answers four questions, which are the ones in the documentation cited above:

• what the tool does, and what it does not do (“changes nothing”);
• when to use it and when to prefer another (in the listing, “to cancel an order use cancel_order”);
• what each parameter means and what changes when you pick one value or another;
• the limits: how many results it returns, what is missing from the response.

The four items have no order, but the second is the most neglected: a model with two tools whose names are close picks badly if the description does not say where one ends and the other begins, and the error does not show up in the schema, it shows up in the agent's choices. For this reason it is worth having test cases in which the right tool is not the most obvious one.

Validating the arguments before executing

The arguments were produced by a model, and there is no guarantee that they respect the schema: they may have the wrong type, out-of-range values or parameters that do not exist. The MCP specification, in its security considerations, says that servers MUST validate all tool inputs, implement access controls, rate limit invocations and sanitize outputs. The listing uses jsonschema with a schema tighter than the one generated before, and tries five calls as a model might produce them.

from jsonschema import Draft202012Validator

SCHEMA = {"type": "object",
          "properties": {"customer": {"type": "string", "minLength": 1, "maxLength": 40},
                         "status": {"enum": ["open", "closed"]},
                         "limit": {"type": "integer", "minimum": 1, "maximum": 50}},
          "required": ["customer"], "additionalProperties": False}
validator = Draft202012Validator(SCHEMA)

CALLS = [   # as a model might produce them: no guarantee on the shape
    {"customer": "Rossi", "status": "open", "limit": 5},
    {"customer": "Rossi", "limit": "five"},
    {"customer": "Rossi", "limit": 5000, "status": "suspended"},
    {"status": "open"},
    {"customer": "Rossi", "order_by": "price"},
]
for number, arguments in enumerate(CALLS, 1):
    errors = sorted(validator.iter_errors(arguments), key=lambda e: list(e.path))
    if not errors:
        print(number, "ok")
    for e in errors:
        print(number, "/".join(map(str, e.path)) or "(root)", "->", e.message)

# Output:
# 1 ok
# 2 limit -> 'five' is not of type 'integer'
# 3 limit -> 5000 is greater than the maximum of 50
# 3 status -> 'suspended' is not one of ['open', 'closed']
# 4 (root) -> 'customer' is a required property
# 5 (root) -> Additional properties are not allowed ('order_by' was unexpected)

Call 1 passes. Call 2 passes a string where an integer is needed; call 3 gets two fields wrong at once and the validator reports both (iter_errors collects all the errors and does not stop at the first, which is valuable, because the model can fix them in one go); call 4 forgets the required field; call 5 invents order_by, and the property is rejected because the schema has additionalProperties: false.

One detail I verified: the validator does not apply default values. With {"type": "integer", "default": 5} and an empty instance it reports no errors and does not insert the 5: in a schema, default is an annotation, so default values are applied by your code after validation.

The database as a tool: parameters, allowed lists, read-only

Databases are the tool that pays off the most and risks the most. The concrete risk is not that the model “turns evil”: it is that the text of a user, or of a document the agent has read, ends up in the parameters and from there in a query.

OWASP, in its guide to preventing SQL injection, names as the preferred defenses parameterized queries for data and, for the structural parts that cannot be parameterized (table and column names, sort direction), a closed list of fixed values decided by the application; it advises against relying on manual escaping. On top of this comes least privilege: the identity that opens the database must not be able to do more than what is needed. The OWASP list on agents with “Excessive Agency” names excessive functions, permissions and autonomy among the causes of harm.

The listing creates a temporary SQLite database and reopens it read-only with the URI ?mode=ro, the way documented by Python. Then it compares a parameterized query with one built as a string, using the same hostile input, x' OR '1'='1, and shows how a column chosen by the model (which cannot be parameterized) is defended with an allowed list.

import os, shutil, sqlite3, tempfile

folder = tempfile.mkdtemp()
path = os.path.join(folder, "shop.db")
writer = sqlite3.connect(path)
writer.execute("CREATE TABLE orders (id INTEGER, customer TEXT, total REAL)")
writer.executemany("INSERT INTO orders VALUES (?, ?, ?)",
                   [(1, "Rossi", 40.0), (2, "Bianchi", 75.5), (3, "Rossi", 12.0)])
writer.commit()
writer.close()

# The tool opens the database read-only: even a wrong query cannot write
db = sqlite3.connect(f"file:{path}?mode=ro", uri=True)

ALLOWED_TABLES = {"orders": ["id", "customer", "total"]}
MAX_LIMIT = 50

def orders_of(customer, limit=5):
    limit = min(int(limit), MAX_LIMIT)
    return db.execute("SELECT id, total FROM orders WHERE customer = ? ORDER BY id LIMIT ?",
                      (customer, limit)).fetchall()

def orders_of_wrong(customer):                # DON'T do this: it concatenates the model's text
    return db.execute(f"SELECT id, total FROM orders WHERE customer = '{customer}'").fetchall()

malicious = "x' OR '1'='1"
print("parameterized, normal input   :", orders_of("Rossi"))
print("parameterized, hostile input  :", orders_of(malicious))
print("concatenated, hostile input   :", orders_of_wrong(malicious))

def order_by(column):                         # column names cannot be parameterized
    if column not in ALLOWED_TABLES["orders"]:
        raise ValueError(f"column not allowed: {column!r}")
    return db.execute(f"SELECT id FROM orders ORDER BY {column} DESC LIMIT 1").fetchall()

for attempt in ("total", "total; DROP TABLE orders"):
    try:
        print("column allowed                :", attempt, "->", order_by(attempt))
    except ValueError as error:
        print("column rejected               :", error)

try:
    db.execute("DELETE FROM orders")
except sqlite3.OperationalError as error:
    print("write on a read-only database :", error)

db.close()
shutil.rmtree(folder)

# Output:
# parameterized, normal input   : [(1, 40.0), (3, 12.0)]
# parameterized, hostile input  : []
# concatenated, hostile input   : [(1, 40.0), (2, 75.5), (3, 12.0)]
# column allowed                : total -> [(2,)]
# column rejected               : column not allowed: 'total; DROP TABLE orders'
# write on a read-only database : attempt to write a readonly database

With the bound parameter, the hostile input is treated as a customer name that does not exist and returns zero rows. With the concatenated string it becomes part of the query and returns all the orders, even those of customers who have nothing to do with it: that is a data leak, not just an error. The column for sorting passes only if it is in the list, and the attempt with ; DROP TABLE is rejected before it reaches the database.

Finally the DELETE fails with attempt to write a readonly database, which is the last-resort protection: if everything else goes wrong, the connection has no permission to write. A caveat on the listing: in SQLite an execute runs a single statement, so the attack with ; DROP would have failed anyway; with other databases and other drivers that is not guaranteed, and the defense must not depend on this property.

The rest of agent security (indirect injection from documents, human confirmation for destructive actions, rate limits) is in the article Security and Safety. Here the principle is enough: an irreversible operation must not be able to start from a single message of the model.

What to send back to the model: the outcome, not the exception

When a tool fails, the model must be able to understand it. The MCP specification distinguishes two cases: protocol errors (unknown tool, malformed request), returned as JSON-RPC errors, and execution errors (an API that does not respond, an invalid input, a violated business rule), reported in the result with isError: true and an actionable text, for example “the departure date must be in the future”. The text of the specification says that the latter allow the model to correct itself and retry, and that clients should pass them on to it.

The listing adopts the same idea with a minimal envelope (ok, and in case of error the kind, the message and retryable). It adds two things that the specification does not decide for you. First, the secret: the token is not a parameter of the tool, so the model does not see it and cannot write it into a prompt; the runtime adds it at call time and the log masks it. Second, the distinction between temporary errors (you retry) and argument errors (retrying is pointless, you send the error back to the model).

SECRETS = {"ORDERS_API": "test_token_12345"}      # in production: a secrets manager, never in the prompt
LOG, FAILED_ATTEMPTS = [], {"order_status": 0}

class TemporaryError(Exception):
    pass

def order_status(order_id: int, token: str):
    """The model does not see `token`: the runtime adds it."""
    if FAILED_ATTEMPTS["order_status"] < 2:            # fails twice, then works
        FAILED_ATTEMPTS["order_status"] += 1
        raise TemporaryError("503 from the orders service")
    if order_id <= 0:
        raise ValueError("order_id must be positive")
    return {"id": order_id, "status": "shipped"}

def outcome_ok(data):
    return {"ok": True, "data": data}

def outcome_error(kind, message, retryable):
    return {"ok": False, "error": {"kind": kind, "message": message, "retryable": retryable}}

def call(function, arguments, maximum=3):
    for attempt in range(1, maximum + 1):
        request = f"{function.__name__}({arguments}, token={SECRETS['ORDERS_API']})"
        LOG.append(request.replace(SECRETS["ORDERS_API"], "***"))     # the log does not contain the secret
        try:
            return outcome_ok(function(**arguments, token=SECRETS["ORDERS_API"]))
        except TemporaryError as e:
            if attempt == maximum:
                return outcome_error("temporary", str(e), True)
        except ValueError as e:                           # the caller's mistake: retrying is useless
            return outcome_error("argument", str(e), False)

print(call(order_status, {"order_id": 7}))
print(call(order_status, {"order_id": -1}))
print(len(LOG), "log lines; the first:", LOG[0])
print("secret in the log:", any(SECRETS["ORDERS_API"] in line for line in LOG))

# Output:
# {'ok': True, 'data': {'id': 7, 'status': 'shipped'}}
# {'ok': False, 'error': {'kind': 'argument', 'message': 'order_id must be positive', 'retryable': False}}
# 4 log lines; the first: order_status({'order_id': 7}, token=***)
# secret in the log: False

The first call fails twice with a temporary error and succeeds on the third; the agent sees only the final outcome. The second has a wrong argument: no new attempt, and the model receives retryable: false with the reason. The four log lines are three attempts plus one, and none contains the token. There is no wait between attempts, on purpose: backoff with jitter and a circuit breaker are in the listing of Multi-Agent Orchestration, and adding them here would only have lengthened the code. For secrets, the perimeter is this: out of the prompt, out of the logs and out of the results returned to the model.

From APIs to tools: generating definitions from OpenAPI

If a service publishes an OpenAPI description, the tool definitions can be derived from the document. The OpenAPI 3.1 specification says that operationId is a string that identifies the operation, that it must be unique among all the operations described, and that “tools and libraries MAY use the operationId to uniquely identify an operation”: it is the natural name of the tool. The listing walks through the paths, turns path and query parameters into schema properties and, as a precaution, imports only the get methods: everything else is listed as skipped.

import json

OPENAPI = {"openapi": "3.1.0", "info": {"title": "Orders", "version": "1.0"},
  "paths": {
    "/orders/{id}": {
      "get": {"operationId": "getOrder", "summary": "Reads an order",
              "parameters": [{"name": "id", "in": "path", "required": True,
                              "schema": {"type": "integer"}}]},
      "delete": {"operationId": "deleteOrder", "summary": "Deletes an order",
                 "parameters": [{"name": "id", "in": "path", "required": True,
                                 "schema": {"type": "integer"}}]}},
    "/orders": {
      "get": {"operationId": "listOrders", "summary": "Lists the orders",
              "parameters": [{"name": "status", "in": "query", "schema": {"type": "string"}},
                             {"name": "limit", "in": "query", "schema": {"type": "integer"}}]},
      "post": {"summary": "Creates an order",
               "requestBody": {"content": {"application/json": {"schema": {
                   "type": "object", "required": ["customer"],
                   "properties": {"customer": {"type": "string"}}}}}}}}}}

def from_openapi(doc, allowed_methods=("get",)):
    tools, skipped = [], []
    for path, operations in doc["paths"].items():
        for method, op in operations.items():
            name = op.get("operationId")
            if method not in allowed_methods:
                skipped.append(f"{method.upper()} {path} (method not allowed)")
                continue
            if not name:
                skipped.append(f"{method.upper()} {path} (no operationId)")
                continue
            properties = {p["name"]: p["schema"] for p in op.get("parameters", [])}
            required = [p["name"] for p in op.get("parameters", []) if p.get("required")]
            tools.append({"name": name, "description": op["summary"],
                          "parameters": {"type": "object", "properties": properties,
                                         "required": required}})
    return tools, skipped

tools, skipped = from_openapi(OPENAPI)
for t in tools:
    print(t["name"], "|", t["description"], "|", json.dumps(t["parameters"]["properties"]),
          "| required:", t["parameters"]["required"])
print("skipped:", skipped)
print("with POST allowed, skipped:", from_openapi(OPENAPI, ("get", "post"))[1])

# Output:
# getOrder | Reads an order | {"id": {"type": "integer"}} | required: ['id']
# listOrders | Lists the orders | {"status": {"type": "string"}, "limit": {"type": "integer"}} | required: []
# skipped: ['DELETE /orders/{id} (method not allowed)', 'POST /orders (method not allowed)']
# with POST allowed, skipped: ['DELETE /orders/{id} (method not allowed)', 'POST /orders (no operationId)']

The two read operations become tools. Deletion and creation do not get in: the rule is to minimize the functions available to the agent, which OWASP names as the first measure against “Excessive Agency”. The second computation, with post allowed, shows another limit: creation has no operationId, so it has no stable name and must be skipped or named by hand.

The listing does not handle request bodies, $ref references, authentication defined in the security scheme, or other formats such as GraphQL, which I did not try. A generated definition remains a draft: the summary description is written for a human reading documentation, not for a model that has to choose, and it almost always needs rewriting.

Long jobs and too many tools

Some operations do not finish in one turn: analysis of an archive, a build, a report. The scheme that holds up is the job with an identifier: the start tool replies immediately with an id, a second tool reads its status, a third cancels it. The listing simulates the job with a generator, a start tool and a check with a cap: after a certain number of checks the agent gives up and says so, instead of waiting forever. The real job would run in a separate process; here each call to work advances by one step.

def long_analysis(file):
    """A long job as a generator: every yield is a step, with a percentage and a message."""
    for step, message in enumerate(("reading", "analysis", "report"), 1):
        yield round(100 * step / 3), f"{message} of {file}"

class Jobs:
    def __init__(self):
        self.jobs, self.status = {}, {}
    def start(self, generator):                  # the tool replies right away with an identifier
        job_id = f"job-{len(self.jobs) + 1}"
        self.jobs[job_id] = generator
        self.status[job_id] = {"status": "running", "percentage": 0, "message": "queued"}
        return job_id
    def work(self, job_id):                      # in production: a separate process
        if self.status[job_id]["status"] != "running":
            return
        try:
            percentage, message = next(self.jobs[job_id])
            self.status[job_id] = {"status": "running", "percentage": percentage, "message": message}
        except StopIteration:
            self.status[job_id]["status"] = "finished"
    def cancel(self, job_id):
        self.status[job_id] = {"status": "cancelled", "percentage": self.status[job_id]["percentage"],
                               "message": "cancelled by the user"}

def wait_for(jobs, job_id, max_checks):
    """The agent polls the status, but with a cap: after N checks it gives up and says so."""
    for n in range(1, max_checks + 1):
        jobs.work(job_id)
        status = jobs.status[job_id]
        print(f"  check {n}: {status['status']}, {status['percentage']}%, {status['message']}")
        if status["status"] != "running":
            return status["status"]
    return "still_running"

jobs = Jobs()
first = jobs.start(long_analysis("log.txt"))
print(first, "->", wait_for(jobs, first, max_checks=6))
second = jobs.start(long_analysis("db.sql"))
print(second, "->", wait_for(jobs, second, max_checks=2))
jobs.cancel(second)
print(second, "after cancelling:", jobs.status[second])

# Output:
#   check 1: running, 33%, reading of log.txt
#   check 2: running, 67%, analysis of log.txt
#   check 3: running, 100%, report of log.txt
#   check 4: finished, 100%, report of log.txt
# job-1 -> finished
#   check 1: running, 33%, reading of db.sql
#   check 2: running, 67%, analysis of db.sql
# job-2 -> still_running
# job-2 after cancelling: {'status': 'cancelled', 'percentage': 67, 'message': 'cancelled by the user'}

The first job reaches 100% at the third check and at the fourth it is finished; the second, with a cap of two checks, stays “still_running” and is cancelled. The design detail that matters is who does the checking: if the model does it, each check is a conversation turn that costs tokens and latency, so it is better for your code to poll the status at intervals and report to the model only the outcome or a meaningful update. I did not measure what it costs, because I did not call models.

The last problem is the number. With dozens of tools, describing them all in every prompt costs tokens and makes the choice harder; Microsoft's guide on orchestration patterns, cited in the previous article, includes “tool overload” among the reasons why a single agent may not be enough.

One answer is to select, for each request, only the relevant tools by searching them by similarity between the request and their description, with the same retrieval technique described in Memory Systems, threshold included. I did not run it here, and it must be measured on your own set of tools: the wrong selection takes away from the model exactly the tool it needed.

Frequently asked questions

What you can do now

You can derive a tool definition from code knowing what is missing (the constraints) and write a description that also says when not to use it. You can validate the arguments produced by a model and apply the default values yourself, open a database with minimal permissions and tell the data from the structure of a query. You can return to the model an outcome that lets it correct itself, keep secrets out of the prompt and out of the logs, generate draft tools from an OpenAPI description by limiting yourself to reading, and manage a long job with an identifier and a cap on checks.

Long jobs with an identifier find their sequel in production deployment, where a task interrupted by a shutdown resumes; and every call to a tool has a price, which the FinOps article turns into a budget per task.