LearnApplied LLM EngineeringFunction Calling & Tool Use

🤖MediumLLM Agents & Tool Use

Function Calling & Tool Use

Build a safe tool-calling runtime that validates model requests, executes controlled actions, feeds observations back, and evaluates complete workflows.

13 min read

Learning path

Step 55 of 155 in the full curriculum

CoT, ToT & Self-Consistency Prompting MCP & Tool Protocol Standards

In the previous lesson, you learned to spend reasoning effort on a decision made from supplied facts. One fact was deliberately missing: the latest carrier scan for order A10234. No amount of careful prompting can invent a trustworthy live scan. Software has to fetch it.

Function calling gives a language model a typed way to request that fetch. The model doesn't run the order API. It proposes an action such as get_order_status(order_id="A10234"); your runtime checks the request, executes an allowed tool, returns the observation, and asks the model to answer from the result.

This boundary is the start of agent engineering. Once an LLM can request reads or writes against real systems, correctness includes parsing, authorization, retries, side effects, latency, and evaluation of the whole trajectory.

A tool call is a request, not an execution

ShopFlow support agent Luna is handling ticket #48291: "My package is late. Where is it?" The assistant needs a live order lookup. A good loop has four visible events:

User asks a question that depends on external state.
Model emits a named action with typed arguments.
Runtime validates and executes that action.
Model receives the tool observation and writes a grounded reply.

Function-calling loop for a delayed parcel where the model proposes get_order_status, the runtime validates and executes it, then the model answers from the observation. — A model proposes the call; application code owns execution and the resulting side effects.

This distinction matters. If the model requests create_refund, it still hasn't refunded anyone. Your application gets a final chance to reject a wrong customer, an ineligible order, a duplicate action, or a write that needs approval.

Define the smallest useful tool contract

A model chooses tools from the definitions you provide. Each definition needs a name, a description that says when to use it, and an argument schema. Keep the schema narrow: if a status lookup only needs an order ID, don't expose refund fields or free-form .

Tool schema for get_order_status showing its purpose, input fields, required order ID, and rejection of additional arguments. — A compact schema narrows what the model may request and makes runtime validation straightforward.

The following logical definition is provider-neutral. Hosted APIs serialize similar information in their own request envelope.

define-order-status-tool.py

TOOL = {
    "name": "get_order_status",
    "description": "Read live shipment status for a customer order. Do not use for refunds.",
    "parameters": {
        "type": "object",
        "properties": {
            "order_id": {"type": "string"},
            "include_tracking": {"type": "boolean"},
        },
        "required": ["order_id"],
        "additionalProperties": False,
    },
}

parameters = TOOL["parameters"]
print(f"tool_name: {TOOL['name']}")
print(f"required: {parameters['required']}")
print(f"accepts_extra_fields: {parameters['additionalProperties']}")

Output

tool_name: get_order_status
required: ['order_id']
accepts_extra_fields: False

A schema is a contract for shape. It helps the model construct a call and helps your runtime reject malformed input. It doesn't prove that the customer owns the order or that an action is permitted.

Parse and validate before dispatch

The model's output crosses a trust boundary. Even when an API offers constrained or strict , your runtime still owns semantic validation and permission checks. Start with the simplest read-only dispatcher: accept one known tool, allow only named fields, and verify field types.

validate-a-read-call.py

class CallRejected(ValueError):
    pass

def validate_status_call(call: dict[str, object]) -> dict[str, object]:
    if call.get("name") != "get_order_status":
        raise CallRejected("unknown tool")
    args = call.get("args")
    if not isinstance(args, dict):
        raise CallRejected("args must be an object")
    allowed = {"order_id", "include_tracking"}
    unknown = set(args) - allowed
    if unknown:
        raise CallRejected(f"unknown fields: {sorted(unknown)}")
    if not isinstance(args.get("order_id"), str):
        raise CallRejected("order_id must be a string")
    if "include_tracking" in args and not isinstance(args["include_tracking"], bool):
        raise CallRejected("include_tracking must be a boolean")
    return args

candidates = [
    {"name": "get_order_status", "args": {"order_id": "A10234", "include_tracking": True}},
    {"name": "get_order_status", "args": {"order_id": "A10234", "refund_now": True}},
]
for call in candidates:
    try:
        args = validate_status_call(call)
        print(f"accepted: {args['order_id']}")
    except CallRejected as exc:
        print(f"rejected: {exc}")

Output

accepted: A10234
rejected: unknown fields: ['refund_now']

Notice that this validator doesn't attempt to repair a bad call silently. A rejected request becomes a structured observation, so the model may correct it on a later bounded turn.

Build the complete tool loop

Function calling becomes concrete only when you run the full state transition. In a hosted-model integration, the first model response contains a tool request and the second model response consumes a tool result. To keep this lab executable without credentials, the model below is scripted while the runtime path is real.

complete-tool-loop.py

import json

ORDERS = {
    "A10234": {"status": "delayed", "carrier": "FastShip", "eta": "Friday"},
}

class ScriptedModel:
    def __init__(self) -> None:
        self.turns = 0

    def respond(self, messages: list[dict[str, object]]) -> dict[str, object]:
        self.turns += 1
        observations = [item for item in messages if item["role"] == "tool"]
        if not observations:
            return {
                "role": "assistant",
                "tool_call": {
                    "id": "status-1",
                    "name": "get_order_status",
                    "args": {"order_id": "A10234"},
                },
            }
        result = json.loads(str(observations[-1]["content"]))
        return {
            "role": "assistant",
            "content": (
                f"Order A10234 is {result['status']} with {result['carrier']} "
                f"and is now expected {result['eta']}."
            ),
        }

def execute_status_tool(call: dict[str, object]) -> dict[str, str]:
    if call.get("name") != "get_order_status":
        raise ValueError("tool not allowed")
    args = call.get("args")
    if not isinstance(args, dict) or set(args) != {"order_id"}:
        raise ValueError("expected only order_id")
    order_id = args["order_id"]
    if not isinstance(order_id, str) or order_id not in ORDERS:
        raise ValueError("unknown order")
    return ORDERS[order_id]

model = ScriptedModel()
messages: list[dict[str, object]] = [
    {"role": "user", "content": "Where is my order A10234?"}
]

first = model.respond(messages)
call = first["tool_call"]
messages.append(first)
observation = execute_status_tool(call)  # runtime executes, not model
messages.append(
    {"role": "tool", "tool_call_id": call["id"], "content": json.dumps(observation)}
)
final = model.respond(messages)

print(f"requested_tool: {call['name']}")
print(f"tool_status: {observation['status']}")
print(f"answer: {final['content']}")
print(f"model_turns: {model.turns}")

Output

requested_tool: get_order_status
tool_status: delayed
answer: Order A10234 is delayed with FastShip and is now expected Friday.
model_turns: 2

The transcript is the essential pattern: user message, assistant tool request, tool observation, assistant answer. Preserve a call ID so each observation stays linked to the request that produced it, especially when multiple reads run concurrently. Toolformer showed that models can learn where API calls help during generation, and ReAct made the reason/action/observation loop explicit for tool-using tasks.^[1]^[2] The engineering burden remains in your runtime.

Structure isn't permission

Valid arguments can still request a harmful or unauthorized action. A status lookup is read-only; create_refund changes customer money. Writes need more checks:

caller owns the target order;
order satisfies refund policy;
amount is computed from trusted order data, not model text;
a human confirms when policy requires approval; and
an key prevents retrying the same write twice.

Refund request checked for schema, ownership and policy, confirmation, and idempotency before any write executes. — A valid function call can reach a write tool only after the runtime validates who may do what, once.

guard-a-refund-write.py

from dataclasses import dataclass

@dataclass(frozen=True)
class Session:
    customer_id: str

ORDERS = {
    "A10234": {"customer_id": "C17", "eligible": True, "paid_cents": 4900},
}
REFUNDS: dict[str, int] = {}

def create_refund(session: Session, order_id: str, confirmed: bool) -> str:
    order = ORDERS.get(order_id)
    if order is None:
        return "blocked: unknown order"
    if session.customer_id != order["customer_id"]:
        return "blocked: order ownership failed"
    if not order["eligible"]:
        return "blocked: refund policy failed"
    if not confirmed:
        return "blocked: confirmation required"
    key = f"refund:{order_id}"
    if key in REFUNDS:
        return f"replayed: refund already exists for {REFUNDS[key]} cents"
    REFUNDS[key] = order["paid_cents"]
    return f"created: refund for {REFUNDS[key]} cents"

print(create_refund(Session("C17"), "Z99999", confirmed=True))
print(create_refund(Session("C99"), "A10234", confirmed=True))
print(create_refund(Session("C17"), "A10234", confirmed=False))
print(create_refund(Session("C17"), "A10234", confirmed=True))
print(create_refund(Session("C17"), "A10234", confirmed=True))

Output

blocked: unknown order
blocked: order ownership failed
blocked: confirmation required
created: refund for 4900 cents
replayed: refund already exists for 4900 cents

This is why schema enforcement isn't a security boundary. It can narrow a JSON shape; only application logic knows ownership, policy, approval, and whether a write already happened.

Return errors as observations, with limits

Tool calls fail in ordinary ways: the model uses an old field name, an order ID doesn't exist, or a service times out. A useful runtime returns a typed rejection rather than a stack trace. The next model turn can correct the request, but it should get only a small retry budget.

correct-a-rejected-call.py

ORDERS = {"A10234": {"status": "delayed"}}

def execute(call: dict[str, object]) -> dict[str, object]:
    if call.get("name") != "get_order_status":
        return {"ok": False, "error": "tool not allowed"}
    args = call.get("args")
    if not isinstance(args, dict):
        return {"ok": False, "error": "args must be an object"}
    unknown = sorted(set(args) - {"order_id"})
    if unknown:
        return {"ok": False, "error": f"unknown fields: {unknown}"}
    if "order_id" not in args:
        return {"ok": False, "error": "required field: order_id"}
    order_id = args["order_id"]
    if not isinstance(order_id, str) or order_id not in ORDERS:
        return {"ok": False, "error": "unknown order"}
    return {"ok": True, "status": ORDERS[order_id]["status"]}

model_attempts = [
    {"name": "get_order_status", "args": {}},
    {"name": "get_order_status", "args": {"order_id": "A10234"}},
]
for turn, call in enumerate(model_attempts, start=1):
    observation = execute(call)
    print(f"turn_{turn}: {observation}")
    if observation["ok"]:
        break

Output

turn_1: {'ok': False, 'error': 'required field: order_id'}
turn_2: {'ok': True, 'status': 'delayed'}

Correction is useful only while it changes the request. If a model repeats the same rejected call, stop before it burns external capacity or triggers repeated writes.

stop-a-repeated-tool-loop.py

import json

attempts = [
    {"name": "get_order_status", "args": {"tracking_id": "A10234"}},
    {"name": "get_order_status", "args": {"tracking_id": "A10234"}},
    {"name": "get_order_status", "args": {"order_id": "A10234"}},
]
seen: set[str] = set()
max_turns = 3

def rejection_for(call: dict[str, object]) -> str | None:
    if call.get("name") != "get_order_status":
        return "tool not allowed"
    args = call.get("args")
    if not isinstance(args, dict):
        return "args must be an object"
    unknown = sorted(set(args) - {"order_id"})
    if unknown:
        return f"unknown fields: {unknown}"
    if "order_id" not in args:
        return "required field: order_id"
    return None

for turn, call in enumerate(attempts[:max_turns], start=1):
    error = rejection_for(call)
    if error is None:
        print(f"turn_{turn}: accepted")
        break
    fingerprint = json.dumps(call, sort_keys=True)
    if fingerprint in seen:
        print(f"turn_{turn}: stopped repeated rejected call")
        break
    seen.add(fingerprint)
    print(f"turn_{turn}: rejected: {error}")

Output

turn_1: rejected: unknown fields: ['tracking_id']
turn_2: stopped repeated rejected call

Track a turn cap, repeated-call detection, timeout budget, and cost budget for each request. A model that can't recover should return a safe fallback or hand the ticket to a person.

Parallelize independent reads, not dependent writes

Alex asks, "Compare my two shipments, A10234 and B77120." Those two status reads are independent. Your runtime may execute them concurrently after validating both calls. A request to cancel one shipment based on those results cannot run at the same time: the write depends on what the reads discover.

parallel-read-only-lookups.py

import asyncio

STATUSES = {
    "A10234": "delayed",
    "B77120": "out_for_delivery",
}

async def get_order_status(order_id: str) -> tuple[str, str]:
    await asyncio.sleep(0)
    return order_id, STATUSES[order_id]

async def main() -> None:
    order_ids = ["A10234", "B77120"]
    rows = await asyncio.gather(*(get_order_status(order_id) for order_id in order_ids))
    for order_id, status in sorted(rows):
        print(f"{order_id}: {status}")
    print("write_action: wait for validated read results")

asyncio.run(main())

Output

A10234: delayed
B77120: out_for_delivery
write_action: wait for validated read results

Concurrency saves elapsed time only when actions are independent. For writes on the same order, serialize execution and apply idempotency rules.

Expose a small allowed toolbox

As an agent grows, passing every internal action to the model wastes context and expands the space of possible mistakes. Filter for authorization first, then route among permitted tools relevant to the current request. Retrieval-augmented API selection appears in Gorilla, which evaluated models against large API collections.^[3]

Tool routing flow where authorization filters a catalog and relevance ranking exposes only get_order_status for a shipment-status question. — The model should choose from a small permitted set, not discover restricted write tools by accident.

route-only-allowed-tools.py

import re
from dataclasses import dataclass

@dataclass(frozen=True)
class Tool:
    name: str
    description: str

catalog = [
    Tool("get_order_status", "shipment status carrier tracking delayed package"),
    Tool("create_refund", "issue refund payment return"),
    Tool("search_policy", "find returns policy eligibility rules"),
]
allowed = {"get_order_status", "search_policy"}
query = "Where is my delayed package shipment?"
query_terms = set(re.findall(r"[a-z_]+", query.lower()))

ranked = sorted(
    (
        (len(query_terms & set(tool.description.split())), tool.name)
        for tool in catalog
        if tool.name in allowed
    ),
    reverse=True,
)
visible_tools = [name for score, name in ranked if score > 0]

print(f"model_visible_tools: {visible_tools}")
print(f"refund_tool_exposed: {'create_refund' in visible_tools}")

Output

model_visible_tools: ['get_order_status']
refund_tool_exposed: False

Routing is not authorization. The allowlist is applied first. A highly relevant but forbidden write tool must stay unavailable.

Evaluate the trajectory, not just the final sentence

A pleasant answer can hide a wrong tool call, an unsafe write, or a result invented without an observation. Evaluate the events your runtime controls:

Check	Passing behavior
Tool choice	Requests `get_order_status` for a live shipment question
Arguments	Uses allowed keys and exact order ID
Execution safety	Makes no unauthorized write
Grounding	Final response reflects returned status
Efficiency	Stays within round, latency, and cost budgets

BFCL evaluates function-selection and executable calling behavior, while Tau-Bench examines longer, policy-constrained interactions with tools and users.^[4]^[5] Your own release gate needs the ShopFlow schemas and policy failures your users will hit.

score-a-tool-trajectory.py

def score(events: list[dict[str, object]]) -> tuple[bool, str]:
    calls = [event for event in events if event["type"] == "call"]
    observations = [event for event in events if event["type"] == "observation"]
    answers = [event for event in events if event["type"] == "answer"]
    if len(calls) != 1 or calls[0].get("name") != "get_order_status":
        return False, "wrong call sequence"
    if calls[0].get("args") != {"order_id": "A10234"}:
        return False, "wrong arguments"
    if not calls[0].get("id") or not observations:
        return False, "observation mismatched call"
    if observations[0].get("call_id") != calls[0]["id"]:
        return False, "observation mismatched call"
    if observations[0].get("status") != "delayed":
        return False, "missing grounded observation"
    if not answers or "delayed" not in str(answers[0].get("text", "")).lower():
        return False, "answer ignored observation"
    return True, "trajectory passed"

good = [
    {
        "type": "call",
        "id": "status-1",
        "name": "get_order_status",
        "args": {"order_id": "A10234"},
    },
    {"type": "observation", "call_id": "status-1", "status": "delayed"},
    {"type": "answer", "text": "Your order is delayed."},
]
bad = [
    {"type": "answer", "text": "Your order should arrive today."},
]
print(score(good))
print(score(bad))

Output

(True, 'trajectory passed')
(False, 'wrong call sequence')

Once correctness is scored, add serving constraints. A runtime that succeeds after ten retries is not ready for a customer-facing workflow.

release-gate-tool-runtime.py

runs = [
    {"passed": True, "unsafe_writes": 0, "rounds": 2, "latency_ms": 430, "cost_cents": 2.1},
    {"passed": True, "unsafe_writes": 0, "rounds": 2, "latency_ms": 510, "cost_cents": 2.4},
    {"passed": False, "unsafe_writes": 0, "rounds": 3, "latency_ms": 680, "cost_cents": 3.8},
    {"passed": True, "unsafe_writes": 0, "rounds": 2, "latency_ms": 470, "cost_cents": 2.2},
]

success_rate = sum(run["passed"] for run in runs) / len(runs)
unsafe_writes = sum(run["unsafe_writes"] for run in runs)
max_rounds = max(run["rounds"] for run in runs)
max_latency_ms = max(run["latency_ms"] for run in runs)
max_cost_cents = max(run["cost_cents"] for run in runs)
latency_budget_ms = 600
cost_budget_cents = 3.0
ready = (
    success_rate >= 0.75
    and unsafe_writes == 0
    and max_rounds <= 3
    and max_latency_ms <= latency_budget_ms
    and max_cost_cents <= cost_budget_cents
)

print(f"success_rate: {success_rate:.0%}")
print(f"unsafe_writes: {unsafe_writes}")
print(f"max_rounds: {max_rounds}")
print(f"max_latency_ms: {max_latency_ms}")
print(f"latency_budget_ms: {latency_budget_ms}")
print(f"max_cost_cents: {max_cost_cents:.1f}")
print(f"cost_budget_cents: {cost_budget_cents:.1f}")
print(f"release_candidate: {ready}")

Output

success_rate: 75%
unsafe_writes: 0
max_rounds: 3
max_latency_ms: 680
latency_budget_ms: 600
max_cost_cents: 3.8
cost_budget_cents: 3.0
release_candidate: False

The failed release is deliberate: one run exceeds latency and cost budgets even though aggregate success reaches the threshold. These fixtures test controller behavior, not model quality. Replace them with held-out tickets, actual model calls, sandboxed tool results, and labeled policy outcomes before shipping.

From local functions to reusable tools

This lesson used in-process Python functions because the execution boundary is easiest to understand there. Real agent products need many capabilities owned by different teams and consumed by more than one host. Rebuilding a custom adapter for every host and service does not scale.

The next lesson introduces the Model Context Protocol (MCP). The mental model stays the same: model proposes a typed action and a trusted runtime decides whether to execute it. MCP standardizes how hosts discover and invoke externally provided capabilities.

What to remember

The model requests; runtime executes. Never give a text generator implicit authority over real effects.
Schemas narrow shape, not policy. Validate ownership, eligibility, approvals, and idempotency before writes.
Observations close the loop. A grounded answer must follow a returned tool result.
Recovery needs budgets. Reject bad calls with structured errors, then cap retries, repeated actions, latency, and cost.
Evaluate traces. Tool choice, arguments, results, safety, and serving cost all belong in the release gate.

Mastery check

Key concepts

Tool call proposal versus application execution
JSON-like input schemas and server-side validation
Assistant request, tool observation, final-answer loop
Read versus write permissions
Confirmation and idempotency for side effects
Structured errors and bounded correction
Parallel safe reads and serialized writes
Allowed-tool routing
Trajectory-based evaluation
Function calling as prerequisite for MCP

Evaluation rubric

Foundational: Explains why a model cannot supply a live carrier scan without a tool observation.
Intermediate: Defines and validates a narrow get_order_status contract.
Intermediate: Implements a complete tool request, execution, observation, and answer loop.
Advanced: Protects a refund write with ownership, policy, confirmation, and idempotency checks.
Advanced: Builds a trajectory release gate with correctness, unsafe-write, round, and latency measures.

Common pitfalls

Treating a tool call as a completed action: The assistant claims a refund happened before execution. Make the runtime return an observation before wording the result as complete.
Trusting schema-valid writes: Correct JSON can still target the wrong order. Apply application authorization and policy checks.
Blind retries: A rejected call repeats until budget is gone. Fingerprint rejected requests and cap tool rounds.
Parallelizing effects: Two write calls race on the same order. Parallelize independent reads only.
Scoring only the final sentence: A correct-looking answer can be ungrounded. Evaluate call, arguments, observation, and outcome.

Practice extension

Extend complete-tool-loop.py with a second read-only tool, get_return_policy(order_id), and a protected write tool, create_return_label(order_id). Build five labeled ticket fixtures: two straightforward status questions, one unknown order, one eligible return, and one ineligible return. Your artifact is a trajectory report containing final outcome, blocked writes, retries, tool rounds, and maximum latency for every fixture.

Next Step

Continue to MCP & Tool Protocol Standards

You can now implement a safe local tool loop and evaluate its trajectories; next you will learn how hosts and external capability servers standardize that same boundary across integrations.

PreviousCoT, ToT & Self-Consistency Prompting

Share this article

X Facebook LinkedIn Bluesky Reddit Hacker News Email

References

Toolformer: Language Models Can Teach Themselves to Use Tools.

Schick, T., et al. · 2023 · NeurIPS 2023

ReAct: Synergizing Reasoning and Acting in Language Models.

Yao, S., Zhao, J., Yu, D., Du, N., Shafran, I., Narasimhan, K., & Cao, Y. · 2023 · ICLR 2023

Gorilla: Large Language Model Connected with Massive APIs.

Patil, S. G., et al. · 2023 · arXiv preprint

Berkeley Function-Calling Leaderboard.

Patil, S. G., et al. · 2024 · UC Berkeley Gorilla repository

Tau-Bench: A Benchmark for Tool-Agent-User Interaction in Real-World Domains

Yao, S., et al. · 2024 · arXiv preprint

Back to Topics

LearnApplied LLM EngineeringFunction Calling & Tool Use

🤖MediumLLM Agents & Tool Use

Function Calling & Tool Use

Build a safe tool-calling runtime that validates model requests, executes controlled actions, feeds observations back, and evaluates complete workflows.

13 min read

Learning path

Step 55 of 155 in the full curriculum

CoT, ToT & Self-Consistency Prompting MCP & Tool Protocol Standards

A tool call is a request, not an execution

ShopFlow support agent Luna is handling ticket #48291: "My package is late. Where is it?" The assistant needs a live order lookup. A good loop has four visible events:

User asks a question that depends on external state.
Model emits a named action with typed arguments.
Runtime validates and executes that action.
Model receives the tool observation and writes a grounded reply.

Define the smallest useful tool contract

The following logical definition is provider-neutral. Hosted APIs serialize similar information in their own request envelope.

define-order-status-tool.py

TOOL = {
    "name": "get_order_status",
    "description": "Read live shipment status for a customer order. Do not use for refunds.",
    "parameters": {
        "type": "object",
        "properties": {
            "order_id": {"type": "string"},
            "include_tracking": {"type": "boolean"},
        },
        "required": ["order_id"],
        "additionalProperties": False,
    },
}

parameters = TOOL["parameters"]
print(f"tool_name: {TOOL['name']}")
print(f"required: {parameters['required']}")
print(f"accepts_extra_fields: {parameters['additionalProperties']}")

Output

tool_name: get_order_status
required: ['order_id']
accepts_extra_fields: False

A schema is a contract for shape. It helps the model construct a call and helps your runtime reject malformed input. It doesn't prove that the customer owns the order or that an action is permitted.

Parse and validate before dispatch

validate-a-read-call.py

class CallRejected(ValueError):
    pass

def validate_status_call(call: dict[str, object]) -> dict[str, object]:
    if call.get("name") != "get_order_status":
        raise CallRejected("unknown tool")
    args = call.get("args")
    if not isinstance(args, dict):
        raise CallRejected("args must be an object")
    allowed = {"order_id", "include_tracking"}
    unknown = set(args) - allowed
    if unknown:
        raise CallRejected(f"unknown fields: {sorted(unknown)}")
    if not isinstance(args.get("order_id"), str):
        raise CallRejected("order_id must be a string")
    if "include_tracking" in args and not isinstance(args["include_tracking"], bool):
        raise CallRejected("include_tracking must be a boolean")
    return args

candidates = [
    {"name": "get_order_status", "args": {"order_id": "A10234", "include_tracking": True}},
    {"name": "get_order_status", "args": {"order_id": "A10234", "refund_now": True}},
]
for call in candidates:
    try:
        args = validate_status_call(call)
        print(f"accepted: {args['order_id']}")
    except CallRejected as exc:
        print(f"rejected: {exc}")

Output

accepted: A10234
rejected: unknown fields: ['refund_now']

Notice that this validator doesn't attempt to repair a bad call silently. A rejected request becomes a structured observation, so the model may correct it on a later bounded turn.

Build the complete tool loop

complete-tool-loop.py

import json

ORDERS = {
    "A10234": {"status": "delayed", "carrier": "FastShip", "eta": "Friday"},
}

class ScriptedModel:
    def __init__(self) -> None:
        self.turns = 0

    def respond(self, messages: list[dict[str, object]]) -> dict[str, object]:
        self.turns += 1
        observations = [item for item in messages if item["role"] == "tool"]
        if not observations:
            return {
                "role": "assistant",
                "tool_call": {
                    "id": "status-1",
                    "name": "get_order_status",
                    "args": {"order_id": "A10234"},
                },
            }
        result = json.loads(str(observations[-1]["content"]))
        return {
            "role": "assistant",
            "content": (
                f"Order A10234 is {result['status']} with {result['carrier']} "
                f"and is now expected {result['eta']}."
            ),
        }

def execute_status_tool(call: dict[str, object]) -> dict[str, str]:
    if call.get("name") != "get_order_status":
        raise ValueError("tool not allowed")
    args = call.get("args")
    if not isinstance(args, dict) or set(args) != {"order_id"}:
        raise ValueError("expected only order_id")
    order_id = args["order_id"]
    if not isinstance(order_id, str) or order_id not in ORDERS:
        raise ValueError("unknown order")
    return ORDERS[order_id]

model = ScriptedModel()
messages: list[dict[str, object]] = [
    {"role": "user", "content": "Where is my order A10234?"}
]

first = model.respond(messages)
call = first["tool_call"]
messages.append(first)
observation = execute_status_tool(call)  # runtime executes, not model
messages.append(
    {"role": "tool", "tool_call_id": call["id"], "content": json.dumps(observation)}
)
final = model.respond(messages)

print(f"requested_tool: {call['name']}")
print(f"tool_status: {observation['status']}")
print(f"answer: {final['content']}")
print(f"model_turns: {model.turns}")

Output

requested_tool: get_order_status
tool_status: delayed
answer: Order A10234 is delayed with FastShip and is now expected Friday.
model_turns: 2

Structure isn't permission

Valid arguments can still request a harmful or unauthorized action. A status lookup is read-only; create_refund changes customer money. Writes need more checks:

caller owns the target order;
order satisfies refund policy;
amount is computed from trusted order data, not model text;
a human confirms when policy requires approval; and
an key prevents retrying the same write twice.

guard-a-refund-write.py

from dataclasses import dataclass

@dataclass(frozen=True)
class Session:
    customer_id: str

ORDERS = {
    "A10234": {"customer_id": "C17", "eligible": True, "paid_cents": 4900},
}
REFUNDS: dict[str, int] = {}

def create_refund(session: Session, order_id: str, confirmed: bool) -> str:
    order = ORDERS.get(order_id)
    if order is None:
        return "blocked: unknown order"
    if session.customer_id != order["customer_id"]:
        return "blocked: order ownership failed"
    if not order["eligible"]:
        return "blocked: refund policy failed"
    if not confirmed:
        return "blocked: confirmation required"
    key = f"refund:{order_id}"
    if key in REFUNDS:
        return f"replayed: refund already exists for {REFUNDS[key]} cents"
    REFUNDS[key] = order["paid_cents"]
    return f"created: refund for {REFUNDS[key]} cents"

print(create_refund(Session("C17"), "Z99999", confirmed=True))
print(create_refund(Session("C99"), "A10234", confirmed=True))
print(create_refund(Session("C17"), "A10234", confirmed=False))
print(create_refund(Session("C17"), "A10234", confirmed=True))
print(create_refund(Session("C17"), "A10234", confirmed=True))

Output

blocked: unknown order
blocked: order ownership failed
blocked: confirmation required
created: refund for 4900 cents
replayed: refund already exists for 4900 cents

This is why schema enforcement isn't a security boundary. It can narrow a JSON shape; only application logic knows ownership, policy, approval, and whether a write already happened.

Return errors as observations, with limits

correct-a-rejected-call.py

ORDERS = {"A10234": {"status": "delayed"}}

def execute(call: dict[str, object]) -> dict[str, object]:
    if call.get("name") != "get_order_status":
        return {"ok": False, "error": "tool not allowed"}
    args = call.get("args")
    if not isinstance(args, dict):
        return {"ok": False, "error": "args must be an object"}
    unknown = sorted(set(args) - {"order_id"})
    if unknown:
        return {"ok": False, "error": f"unknown fields: {unknown}"}
    if "order_id" not in args:
        return {"ok": False, "error": "required field: order_id"}
    order_id = args["order_id"]
    if not isinstance(order_id, str) or order_id not in ORDERS:
        return {"ok": False, "error": "unknown order"}
    return {"ok": True, "status": ORDERS[order_id]["status"]}

model_attempts = [
    {"name": "get_order_status", "args": {}},
    {"name": "get_order_status", "args": {"order_id": "A10234"}},
]
for turn, call in enumerate(model_attempts, start=1):
    observation = execute(call)
    print(f"turn_{turn}: {observation}")
    if observation["ok"]:
        break

Output

turn_1: {'ok': False, 'error': 'required field: order_id'}
turn_2: {'ok': True, 'status': 'delayed'}

Correction is useful only while it changes the request. If a model repeats the same rejected call, stop before it burns external capacity or triggers repeated writes.

stop-a-repeated-tool-loop.py

import json

attempts = [
    {"name": "get_order_status", "args": {"tracking_id": "A10234"}},
    {"name": "get_order_status", "args": {"tracking_id": "A10234"}},
    {"name": "get_order_status", "args": {"order_id": "A10234"}},
]
seen: set[str] = set()
max_turns = 3

def rejection_for(call: dict[str, object]) -> str | None:
    if call.get("name") != "get_order_status":
        return "tool not allowed"
    args = call.get("args")
    if not isinstance(args, dict):
        return "args must be an object"
    unknown = sorted(set(args) - {"order_id"})
    if unknown:
        return f"unknown fields: {unknown}"
    if "order_id" not in args:
        return "required field: order_id"
    return None

for turn, call in enumerate(attempts[:max_turns], start=1):
    error = rejection_for(call)
    if error is None:
        print(f"turn_{turn}: accepted")
        break
    fingerprint = json.dumps(call, sort_keys=True)
    if fingerprint in seen:
        print(f"turn_{turn}: stopped repeated rejected call")
        break
    seen.add(fingerprint)
    print(f"turn_{turn}: rejected: {error}")

Output

turn_1: rejected: unknown fields: ['tracking_id']
turn_2: stopped repeated rejected call

Track a turn cap, repeated-call detection, timeout budget, and cost budget for each request. A model that can't recover should return a safe fallback or hand the ticket to a person.

Parallelize independent reads, not dependent writes

parallel-read-only-lookups.py

import asyncio

STATUSES = {
    "A10234": "delayed",
    "B77120": "out_for_delivery",
}

async def get_order_status(order_id: str) -> tuple[str, str]:
    await asyncio.sleep(0)
    return order_id, STATUSES[order_id]

async def main() -> None:
    order_ids = ["A10234", "B77120"]
    rows = await asyncio.gather(*(get_order_status(order_id) for order_id in order_ids))
    for order_id, status in sorted(rows):
        print(f"{order_id}: {status}")
    print("write_action: wait for validated read results")

asyncio.run(main())

Output

A10234: delayed
B77120: out_for_delivery
write_action: wait for validated read results

Concurrency saves elapsed time only when actions are independent. For writes on the same order, serialize execution and apply idempotency rules.

Expose a small allowed toolbox

route-only-allowed-tools.py

import re
from dataclasses import dataclass

@dataclass(frozen=True)
class Tool:
    name: str
    description: str

catalog = [
    Tool("get_order_status", "shipment status carrier tracking delayed package"),
    Tool("create_refund", "issue refund payment return"),
    Tool("search_policy", "find returns policy eligibility rules"),
]
allowed = {"get_order_status", "search_policy"}
query = "Where is my delayed package shipment?"
query_terms = set(re.findall(r"[a-z_]+", query.lower()))

ranked = sorted(
    (
        (len(query_terms & set(tool.description.split())), tool.name)
        for tool in catalog
        if tool.name in allowed
    ),
    reverse=True,
)
visible_tools = [name for score, name in ranked if score > 0]

print(f"model_visible_tools: {visible_tools}")
print(f"refund_tool_exposed: {'create_refund' in visible_tools}")

Output

model_visible_tools: ['get_order_status']
refund_tool_exposed: False

Routing is not authorization. The allowlist is applied first. A highly relevant but forbidden write tool must stay unavailable.

Evaluate the trajectory, not just the final sentence

A pleasant answer can hide a wrong tool call, an unsafe write, or a result invented without an observation. Evaluate the events your runtime controls:

Check	Passing behavior
Tool choice	Requests `get_order_status` for a live shipment question
Arguments	Uses allowed keys and exact order ID
Execution safety	Makes no unauthorized write
Grounding	Final response reflects returned status
Efficiency	Stays within round, latency, and cost budgets

score-a-tool-trajectory.py

def score(events: list[dict[str, object]]) -> tuple[bool, str]:
    calls = [event for event in events if event["type"] == "call"]
    observations = [event for event in events if event["type"] == "observation"]
    answers = [event for event in events if event["type"] == "answer"]
    if len(calls) != 1 or calls[0].get("name") != "get_order_status":
        return False, "wrong call sequence"
    if calls[0].get("args") != {"order_id": "A10234"}:
        return False, "wrong arguments"
    if not calls[0].get("id") or not observations:
        return False, "observation mismatched call"
    if observations[0].get("call_id") != calls[0]["id"]:
        return False, "observation mismatched call"
    if observations[0].get("status") != "delayed":
        return False, "missing grounded observation"
    if not answers or "delayed" not in str(answers[0].get("text", "")).lower():
        return False, "answer ignored observation"
    return True, "trajectory passed"

good = [
    {
        "type": "call",
        "id": "status-1",
        "name": "get_order_status",
        "args": {"order_id": "A10234"},
    },
    {"type": "observation", "call_id": "status-1", "status": "delayed"},
    {"type": "answer", "text": "Your order is delayed."},
]
bad = [
    {"type": "answer", "text": "Your order should arrive today."},
]
print(score(good))
print(score(bad))

Output

(True, 'trajectory passed')
(False, 'wrong call sequence')

Once correctness is scored, add serving constraints. A runtime that succeeds after ten retries is not ready for a customer-facing workflow.

release-gate-tool-runtime.py

runs = [
    {"passed": True, "unsafe_writes": 0, "rounds": 2, "latency_ms": 430, "cost_cents": 2.1},
    {"passed": True, "unsafe_writes": 0, "rounds": 2, "latency_ms": 510, "cost_cents": 2.4},
    {"passed": False, "unsafe_writes": 0, "rounds": 3, "latency_ms": 680, "cost_cents": 3.8},
    {"passed": True, "unsafe_writes": 0, "rounds": 2, "latency_ms": 470, "cost_cents": 2.2},
]

success_rate = sum(run["passed"] for run in runs) / len(runs)
unsafe_writes = sum(run["unsafe_writes"] for run in runs)
max_rounds = max(run["rounds"] for run in runs)
max_latency_ms = max(run["latency_ms"] for run in runs)
max_cost_cents = max(run["cost_cents"] for run in runs)
latency_budget_ms = 600
cost_budget_cents = 3.0
ready = (
    success_rate >= 0.75
    and unsafe_writes == 0
    and max_rounds <= 3
    and max_latency_ms <= latency_budget_ms
    and max_cost_cents <= cost_budget_cents
)

print(f"success_rate: {success_rate:.0%}")
print(f"unsafe_writes: {unsafe_writes}")
print(f"max_rounds: {max_rounds}")
print(f"max_latency_ms: {max_latency_ms}")
print(f"latency_budget_ms: {latency_budget_ms}")
print(f"max_cost_cents: {max_cost_cents:.1f}")
print(f"cost_budget_cents: {cost_budget_cents:.1f}")
print(f"release_candidate: {ready}")

Output

success_rate: 75%
unsafe_writes: 0
max_rounds: 3
max_latency_ms: 680
latency_budget_ms: 600
max_cost_cents: 3.8
cost_budget_cents: 3.0
release_candidate: False

From local functions to reusable tools

What to remember

The model requests; runtime executes. Never give a text generator implicit authority over real effects.
Schemas narrow shape, not policy. Validate ownership, eligibility, approvals, and idempotency before writes.
Observations close the loop. A grounded answer must follow a returned tool result.
Recovery needs budgets. Reject bad calls with structured errors, then cap retries, repeated actions, latency, and cost.
Evaluate traces. Tool choice, arguments, results, safety, and serving cost all belong in the release gate.

Mastery check

Key concepts

Tool call proposal versus application execution
JSON-like input schemas and server-side validation
Assistant request, tool observation, final-answer loop
Read versus write permissions
Confirmation and idempotency for side effects
Structured errors and bounded correction
Parallel safe reads and serialized writes
Allowed-tool routing
Trajectory-based evaluation
Function calling as prerequisite for MCP

Evaluation rubric

Foundational: Explains why a model cannot supply a live carrier scan without a tool observation.
Intermediate: Defines and validates a narrow get_order_status contract.
Intermediate: Implements a complete tool request, execution, observation, and answer loop.
Advanced: Protects a refund write with ownership, policy, confirmation, and idempotency checks.
Advanced: Builds a trajectory release gate with correctness, unsafe-write, round, and latency measures.

Common pitfalls

Treating a tool call as a completed action: The assistant claims a refund happened before execution. Make the runtime return an observation before wording the result as complete.
Trusting schema-valid writes: Correct JSON can still target the wrong order. Apply application authorization and policy checks.
Blind retries: A rejected call repeats until budget is gone. Fingerprint rejected requests and cap tool rounds.
Parallelizing effects: Two write calls race on the same order. Parallelize independent reads only.
Scoring only the final sentence: A correct-looking answer can be ungrounded. Evaluate call, arguments, observation, and outcome.

Practice extension

Next Step

Continue to MCP & Tool Protocol Standards

You can now implement a safe local tool loop and evaluate its trajectories; next you will learn how hosts and external capability servers standardize that same boundary across integrations.

PreviousCoT, ToT & Self-Consistency Prompting

Share this article

X Facebook LinkedIn Bluesky Reddit Hacker News Email

References

Toolformer: Language Models Can Teach Themselves to Use Tools.

Schick, T., et al. · 2023 · NeurIPS 2023

ReAct: Synergizing Reasoning and Acting in Language Models.

Yao, S., Zhao, J., Yu, D., Du, N., Shafran, I., Narasimhan, K., & Cao, Y. · 2023 · ICLR 2023

Gorilla: Large Language Model Connected with Massive APIs.

Patil, S. G., et al. · 2023 · arXiv preprint

Berkeley Function-Calling Leaderboard.

Patil, S. G., et al. · 2024 · UC Berkeley Gorilla repository

Tau-Bench: A Benchmark for Tool-Agent-User Interaction in Real-World Domains

Yao, S., et al. · 2024 · arXiv preprint

Function Calling & Tool Use

A tool call is a request, not an execution

Which component actually touches the order service after an LLM emits get_order_status(order_id="A10234")?

Define the smallest useful tool contract

Parse and validate before dispatch

Build the complete tool loop

Why does a one-tool answer normally require two model turns?

Structure isn't permission

Return errors as observations, with limits

Why shouldn't a runtime retry every failed tool request until the model eventually succeeds?

Parallelize independent reads, not dependent writes

Expose a small allowed toolbox

Evaluate the trajectory, not just the final sentence

From local functions to reusable tools

What to remember

Mastery check

Key concepts

Evaluation rubric

Common pitfalls

Practice extension

What is the central safety rule for a tool-using LLM application?

Function Calling & Tool Use

A tool call is a request, not an execution

Which component actually touches the order service after an LLM emits get_order_status(order_id="A10234")?

Define the smallest useful tool contract

Parse and validate before dispatch

Build the complete tool loop

Why does a one-tool answer normally require two model turns?

Structure isn't permission

Return errors as observations, with limits

Why shouldn't a runtime retry every failed tool request until the model eventually succeeds?

Parallelize independent reads, not dependent writes

Expose a small allowed toolbox

Evaluate the trajectory, not just the final sentence

From local functions to reusable tools

What to remember

Mastery check

Key concepts

Evaluation rubric

Common pitfalls

Practice extension

What is the central safety rule for a tool-using LLM application?