LLM Tool Use and Function Calling

Real-World Scenario

A support bot needs to answer “where is my order?” from the orders database, check a courier’s API, and issue small refunds. The model can’t reach any of that. What it can do is ask your code to call a function, with arguments, and then answer from what the function returned. Everything that makes this safe (checking the arguments, handling failures, deciding which calls need a human) is ordinary code on your side of the API.

The examples run against ScriptedModel, a stand-in for a tool-calling LLM API. It replays replies you script, in the same tool_use and tool_result message shapes that hosted APIs use, and it rejects malformed conversations the way a real API does. No API key is needed. The loop, validation and error handling you write here work unchanged against a real SDK client.

Defining a Tool and Validating Its Arguments

A tool is a name, a description the model reads, and a JSON Schema for its arguments. The model fills in the arguments, so check them before running anything.

import json, sqlite3
from jsonschema import validate, ValidationError

db = sqlite3.connect(":memory:")
db.execute("CREATE TABLE orders (id TEXT PRIMARY KEY, status TEXT, total REAL)")
db.executemany("INSERT INTO orders VALUES (?, ?, ?)",
               [("A1042", "shipped", 59.90), ("A1043", "processing", 12.50)])

TOOLS = [{"name": "get_order",
          "description": "Look up one order by ID, e.g. 'A1042'. Use for any question about an order.",
          "input_schema": {"type": "object", "additionalProperties": False, "required": ["order_id"],
                           "properties": {"order_id": {"type": "string", "pattern": "^A[0-9]{4}$"}}}}]

def get_order(order_id):
    row = db.execute("SELECT status, total FROM orders WHERE id = ?", (order_id,)).fetchone()
    if row is None:
        raise LookupError(f"no order with id {order_id}")
    return {"order_id": order_id, "status": row[0], "total": row[1]}

FUNCTIONS = {"get_order": get_order}

def run_tool(name, args):
    """Validate the model's arguments, run the function, return (content, is_error)."""
    schema = next(t["input_schema"] for t in TOOLS if t["name"] == name)
    try:
        validate(args, schema)
        return json.dumps(FUNCTIONS[name](**args)), False
    except (ValidationError, LookupError) as e:
        return f"error: {getattr(e, 'message', e)}", True

if __name__ == "__main__":
    for args in ({"order_id": "A1042"}, {"order_id": 1042}, {"id": "A1042"}, {"order_id": "A9999"}):
        print(f"{str(args):22} -> {run_tool('get_order', args)}")

$ python tools.py
{'order_id': 'A1042'}  -> ('{"order_id": "A1042", "status": "shipped", "total": 59.9}', False)
{'order_id': 1042}     -> ("error: 1042 is not of type 'string'", True)
{'id': 'A1042'}        -> ("error: Additional properties are not allowed ('id' was unexpected)", True)
{'order_id': 'A9999'}  -> ('error: no order with id A9999', True)

The first call is valid and returns the order. The other three are the ways models really get arguments wrong: a number where the schema says string, an invented parameter name, and a well-formed ID that doesn’t exist. None of them crash. Each becomes a short error string with is_error set to True, which is what gets sent back to the model.

Two details in the schema do real work. "additionalProperties": False is what catches the invented id key; without it {"id": "A1042"} fails only on the missing order_id, and a schema without required would pass it straight into get_order. The pattern keeps free text out of the SQL parameter. The description matters as much as the schema: it is the only thing the model reads when deciding whether to call the tool. Save this file as tools.py.

The Protocol: Every tool_use Needs a tool_result

The stand-in model enforces the one rule hosted APIs are strictest about. Save it as model.py, then send a conversation that breaks the rule.

class BadRequestError(Exception): status = 400

def text(s):
    return {"type": "text", "text": s}

def call(id, name, **args):
    return {"type": "tool_use", "id": id, "name": name, "input": args}

class ScriptedModel:
    """Stands in for a tool-calling LLM API: replays scripted replies, enforces the protocol."""
    def __init__(self, replies):
        self.replies, self.calls = list(replies), 0

    def create(self, messages, tools):
        self.calls += 1
        for i, msg in enumerate(messages):          # every tool_use needs a tool_result next
            ids = {b["id"] for b in msg["content"] if isinstance(b, dict) and b["type"] == "tool_use"}
            nxt = messages[i + 1]["content"] if i + 1 < len(messages) else []
            answered = {b["tool_use_id"] for b in nxt if isinstance(b, dict) and b["type"] == "tool_result"}
            if ids - answered:
                raise BadRequestError(f"messages.{i}: tool_use ids without tool_result "
                                      f"in the next message: {sorted(ids - answered)}")
        content = self.replies.pop(0)
        stop = "tool_use" if any(b["type"] == "tool_use" for b in content) else "end_turn"
        return {"stop_reason": stop, "content": content}

from model import ScriptedModel, text, call
from tools import TOOLS

model = ScriptedModel([
    [text("Let me look that up."), call("toolu_01", "get_order", order_id="A1042")],
    [text("Your order has shipped.")],
])
messages = [{"role": "user", "content": "Where is order A1042?"}]
reply = model.create(messages, TOOLS)
print(reply["stop_reason"], reply["content"][1])

messages.append({"role": "assistant", "content": reply["content"]})
messages.append({"role": "user", "content": "ok, and?"})   # forgot to run the tool
model.create(messages, TOOLS)

$ python broken.py
tool_use {'type': 'tool_use', 'id': 'toolu_01', 'name': 'get_order', 'input': {'order_id': 'A1042'}}
Traceback (most recent call last):
  File "/home/claude/llm-demo/tool-use/broken.py", line 14, in <module>
    model.create(messages, TOOLS)
  File "/home/claude/llm-demo/tool-use/model.py", line 21, in create
    raise BadRequestError(f"messages.{i}: tool_use ids without tool_result "
model.BadRequestError: messages.1: tool_use ids without tool_result in the next message: ['toolu_01']

The first reply has stop_reason tool_use: the model has stopped and is waiting for your code. The second request answered that with an ordinary user message instead of the tool’s result, and the API rejected the whole conversation with a 400. Real APIs refuse this for the same reason: the model was promised a result for toolu_01 and never got one. The fix is always the same: after a tool_use reply, the next message must carry a tool_result for every tool_use id in it.

The Tool-Use Loop

The loop calls the model, runs whatever tools it asks for, sends the results back, and repeats until the model answers in text.

from tools import TOOLS, run_tool

def run(model, question, max_turns=5):
    messages = [{"role": "user", "content": question}]
    for _ in range(max_turns):
        reply = model.create(messages, TOOLS)
        messages.append({"role": "assistant", "content": reply["content"]})
        if reply["stop_reason"] != "tool_use":
            return reply["content"][-1]["text"]
        results = []
        for block in reply["content"]:
            if block["type"] == "tool_use":
                content, is_error = run_tool(block["name"], block["input"])
                print(f"  {block['id']} {block['name']}({block['input']}) -> {content}")
                results.append({"type": "tool_result", "tool_use_id": block["id"],
                                "content": content, "is_error": is_error})
        messages.append({"role": "user", "content": results})   # all results, one message
    return f"[stopped after {max_turns} model calls]"

if __name__ == "__main__":
    from model import ScriptedModel, text, call
    model = ScriptedModel([
        [text("Let me look that up."), call("toolu_01", "get_order", order_id="A1042")],
        [text("Order A1042 has shipped. The total was $59.90.")],
    ])
    print(run(model, "Where is order A1042?"), f"({model.calls} model calls)")

$ python agent.py
  toolu_01 get_order({'order_id': 'A1042'}) -> {"order_id": "A1042", "status": "shipped", "total": 59.9}
Order A1042 has shipped. The total was $59.90. (2 model calls)

It took two model calls for one question. The first returned the tool_use block, the loop ran get_order and sent the JSON back, and the second produced the answer from that JSON. The $59.90 in the reply came from the database via the tool result, not from the model’s training data. That is the whole point of tool use.

max_turns counts model calls, not tools. It is the loop’s only guarantee that it finishes. Save this as agent.py.

When Tools Fail

Error results let the model correct itself. The same errors from tools.py, now inside the loop:

from agent import run
from model import ScriptedModel, text, call

model = ScriptedModel([
    [call("toolu_01", "get_order", order_id=1042)],            # wrong type
    [call("toolu_02", "get_order", order_id="A1042")],         # model reads the error, retries
    [text("Order A1042 has shipped.")],
])
print(run(model, "status of order 1042?"), f"({model.calls} model calls)\n")

model = ScriptedModel([
    [call("toolu_03", "get_order", order_id="A9999")],
    [text("I couldn't find order A9999. Could you check the number on your receipt?")],
])
print(run(model, "where is A9999?"), f"({model.calls} model calls)\n")

model = ScriptedModel([[call(f"toolu_{i}", "get_order", order_id="A1043")] for i in range(9)])
print(run(model, "is A1043 done yet?"), f"({model.calls} model calls)")

$ python recover.py
  toolu_01 get_order({'order_id': 1042}) -> error: 1042 is not of type 'string'
  toolu_02 get_order({'order_id': 'A1042'}) -> {"order_id": "A1042", "status": "shipped", "total": 59.9}
Order A1042 has shipped. (3 model calls)

  toolu_03 get_order({'order_id': 'A9999'}) -> error: no order with id A9999
I couldn't find order A9999. Could you check the number on your receipt? (2 model calls)

  toolu_0 get_order({'order_id': 'A1043'}) -> {"order_id": "A1043", "status": "processing", "total": 12.5}
  toolu_1 get_order({'order_id': 'A1043'}) -> {"order_id": "A1043", "status": "processing", "total": 12.5}
  toolu_2 get_order({'order_id': 'A1043'}) -> {"order_id": "A1043", "status": "processing", "total": 12.5}
  toolu_3 get_order({'order_id': 'A1043'}) -> {"order_id": "A1043", "status": "processing", "total": 12.5}
  toolu_4 get_order({'order_id': 'A1043'}) -> {"order_id": "A1043", "status": "processing", "total": 12.5}
[stopped after 5 model calls] (5 model calls)

The first conversation recovers: the type error goes back to the model, which retries with "A1042" and answers correctly. The scripted model stands in for what capable models typically do with a specific error message. The second conversation turns a missing record into a helpful reply instead of an invented status. The third is the failure mode to plan for: a model that keeps calling the same tool. Every call returned the same row, and only max_turns stopped it after 5 model calls. Without the cap this request would run, and bill, until something else timed out.

Running Independent Calls in Parallel

A single model reply can ask for several tools at once. When the calls don’t depend on each other, run them concurrently.

import time
from concurrent.futures import ThreadPoolExecutor
from model import call

def track_parcel(order_id):          # stand-ins for two slow HTTP APIs
    time.sleep(0.4); return f"{order_id}: out for delivery"
def refund_policy(order_id):
    time.sleep(0.4); return f"{order_id}: refundable until 2026-10-12"
SLOW = {"track_parcel": track_parcel, "refund_policy": refund_policy}

blocks = [call("toolu_01", "track_parcel", order_id="A1042"),   # one model reply,
          call("toolu_02", "refund_policy", order_id="A1042")]  # two independent calls

def execute(block):
    return {"type": "tool_result", "tool_use_id": block["id"],
            "content": SLOW[block["name"]](**block["input"])}

if __name__ == "__main__":
    for label, run in [("one by one", lambda: [execute(b) for b in blocks]),
                       ("threads   ", lambda: list(ThreadPoolExecutor().map(execute, blocks)))]:
        t = time.perf_counter()
        results = run()
        print(f"{label} {time.perf_counter() - t:.2f}s", [r["tool_use_id"] for r in results])
    print(results[1]["content"])

$ python parallel.py
one by one 0.80s ['toolu_01', 'toolu_02']
threads    0.40s ['toolu_01', 'toolu_02']
A1042: refundable until 2026-10-12

Two 0.4-second calls take 0.80s one after the other and 0.40s in threads. With five independent lookups the difference is two seconds of a user waiting. ThreadPoolExecutor.map returns results in input order, so the tool_use_id pairing is preserved, and all results go back to the model together in one user message, as the protocol above requires.

Threads suit tools that wait on the network. If one of these tools used the SQLite connection from tools.py, it would fail: a sqlite3 connection may only be used by the thread that created it, unless you open it with check_same_thread=False or give each thread its own connection.

Gating Tools That Change Things

Reading an order is harmless. Refunding one moves money, so the code, not the model, decides whether it happens.

from model import call

ledger = []                                   # what actually left the building

def issue_refund(order_id, amount):
    ledger.append((order_id, amount))
    return f"refunded ${amount:.2f} on {order_id}"

def gated_refund(block, approve):
    """Side-effecting tool: small refunds run, larger ones need a person to say yes."""
    args = block["input"]
    if args["amount"] > 20 and not approve(args):
        return {"type": "tool_result", "tool_use_id": block["id"], "is_error": True,
                "content": "refund over $20 needs human approval; ticket opened, tell the user"}
    return {"type": "tool_result", "tool_use_id": block["id"], "content": issue_refund(**args)}

if __name__ == "__main__":
    reviewer_says_no = lambda args: False
    for block in [call("toolu_01", "issue_refund", order_id="A1043", amount=12.50),
                  call("toolu_02", "issue_refund", order_id="A1042", amount=59.90),
                  call("toolu_03", "issue_refund", order_id="A1042", amount=5990.0)]:
        print(gated_refund(block, reviewer_says_no))
    print("ledger:", ledger)

$ python approval.py
{'type': 'tool_result', 'tool_use_id': 'toolu_01', 'content': 'refunded $12.50 on A1043'}
{'type': 'tool_result', 'tool_use_id': 'toolu_02', 'is_error': True, 'content': 'refund over $20 needs human approval; ticket opened, tell the user'}
{'type': 'tool_result', 'tool_use_id': 'toolu_03', 'is_error': True, 'content': 'refund over $20 needs human approval; ticket opened, tell the user'}
ledger: [('A1043', 12.5)]

The small refund runs. The two larger ones never reach issue_refund: the reviewer said no, so each became an error result telling the model to let the user know a ticket was opened. The ledger shows exactly one refund left the building. The third request asked to refund $5,990 on a $59.90 order, which is what a model misreading a price or following an injected instruction looks like. The approval gate caught it only because it was over $20. Practice task 4 adds the check that should have caught it on its own.

Write this kind of rule in code, never only in the system prompt. A prompt instruction such as “never refund more than $20 without approval” is a request to the model. The if statement in gated_refund is a guarantee.

Practice

1. run_tool("delete_everything", {}) crashes with StopIteration, because the tool lookup finds nothing. Models do sometimes call tools that don’t exist. Wrap run_tool so an unknown name becomes an error result that lists the real tools.

   Solution

   import tools
   
   def run_tool(name, args):
       if name not in tools.FUNCTIONS:
           known = ", ".join(tools.FUNCTIONS)
           return f"error: unknown tool {name!r}; available tools: {known}", True
       return tools.run_tool(name, args)
   
   print(run_tool("delete_everything", {}))
   print(run_tool("get_order", {"order_id": "A1043"}))

   $ python sol_unknown.py
   ("error: unknown tool 'delete_everything'; available tools: get_order", True)
   ('{"order_id": "A1043", "status": "processing", "total": 12.5}', False)

   Listing the available tools in the error gives the model something to correct toward. Check the name before touching the schema, so an unknown name can never reach FUNCTIONS[name].

2. In recover.py’s third conversation the loop ran get_order five times with identical arguments. Change the loop in agent.py to stop as soon as the model repeats an identical call.

   Solution

   Add import json and create seen = set() next to messages at the top of run(). It holds (name, arguments) pairs, with the arguments serialized with sort_keys=True so key order doesn’t matter. Check it before running each tool:

   key = (block["name"], json.dumps(block["input"], sort_keys=True))
   if key in seen:
       return f"[stopped: repeated {block['name']}({block['input']})]"
   seen.add(key)

   $ python sol_repeat.py
     toolu_0 get_order({'order_id': 'A1043'}) -> {"order_id": "A1043", "status": "processing", "total": 12.5}
   [stopped: repeated get_order({'order_id': 'A1043'})] (2 model calls)

   The loop now stops after 2 model calls instead of 5, with a reason you can log. Keep max_turns as well: a model can loop with arguments that differ slightly each time, and only the cap catches that.

3. In parallel.py, make track_parcel raise TimeoutError. ThreadPoolExecutor.map re-raises the exception when you read the results, and the refund-policy result is lost with it. Change execute so a failing tool becomes an error result and the other call’s result still gets through.

   Solution

   from concurrent.futures import ThreadPoolExecutor
   from parallel import SLOW, blocks
   
   def carrier_down(order_id):
       raise TimeoutError("carrier API timed out after 10s")
   SLOW["track_parcel"] = carrier_down
   
   def execute(block):
       try:
           content, is_error = SLOW[block["name"]](**block["input"]), False
       except Exception as e:
           content, is_error = f"error: {type(e).__name__}: {e}", True
       return {"type": "tool_result", "tool_use_id": block["id"], "content": content, "is_error": is_error}
   
   for r in ThreadPoolExecutor().map(execute, blocks):
       print(r)

   $ python sol_parallel_error.py
   {'type': 'tool_result', 'tool_use_id': 'toolu_01', 'content': 'error: TimeoutError: carrier API timed out after 10s', 'is_error': True}
   {'type': 'tool_result', 'tool_use_id': 'toolu_02', 'content': 'A1042: refundable until 2026-10-12', 'is_error': False}

   Both ids still get a tool_result, which the protocol requires, and the model learns that the courier lookup failed while still being able to answer the refund question. Catching broad Exception is right here: this is the boundary where any tool failure has to become data for the model.

4. gated_refund approved nothing over $20, so the $5,990 request was blocked only by luck. With an approver that says yes, it would have gone through. Reject any refund larger than the order’s total before the approval step.

   Solution

   from approval import gated_refund, call
   from tools import get_order
   
   def checked_refund(block, approve):
       order = get_order(block["input"]["order_id"])
       if block["input"]["amount"] > order["total"]:
           return {"type": "tool_result", "tool_use_id": block["id"], "is_error": True,
                   "content": f"refund exceeds order total ${order['total']:.2f}"}
       return gated_refund(block, approve)
   
   yes = lambda args: True
   print(checked_refund(call("toolu_04", "issue_refund", order_id="A1042", amount=5990.0), yes))
   print(checked_refund(call("toolu_05", "issue_refund", order_id="A1042", amount=59.90), yes))

   $ python sol_total.py
   {'type': 'tool_result', 'tool_use_id': 'toolu_04', 'is_error': True, 'content': 'refund exceeds order total $59.90'}
   {'type': 'tool_result', 'tool_use_id': 'toolu_05', 'content': 'refunded $59.90 on A1042'}

   With an approver that says yes to everything, the $5,990 request is still refused, because it fails a check that doesn’t depend on anyone’s judgment. The genuine full refund goes through. Validate against your own records (the order total, the customer on the order, whether it was already refunded) rather than trusting any number the model supplies.