Build a safe tool-calling runtime that validates model requests, executes controlled actions, feeds observations back, and evaluates complete workflows.
Careful reasoning over supplied facts still couldn't answer eval run R42: the latest result was missing. No amount of careful prompting can invent a trustworthy live metric. Software has to fetch it.
Function calling gives a language model a typed way to request that fetch. The model doesn't run the evaluation service. It proposes an action such as get_eval_run(run_id="R42"); your runtime checks the request, executes an allowed tool, returns the observation, and asks the model to answer from the result.
This permission 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.
Release assistant Luna is answering, "Why did eval run R42 fail?" The assistant needs a live eval lookup. A good loop has four visible events:
This distinction matters. If the model requests promote_model, it still hasn't changed production traffic. Your application gets a final chance to reject the wrong run, a failed gate, a duplicate action, or a write that needs approval.
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 an eval lookup only needs a run ID, don't expose promotion fields or free-form SQL.
This logical definition is provider-neutral. Hosted APIs serialize similar information in their own request envelope. If you adapt it to OpenAI strict mode, list every property in required, represent optional values with a nullable type, and keep additionalProperties: false; provider schema subsets differ.[1]
1TOOL = {
2 "name": "get_eval_run",
3 "description": "Read live evaluation status for one model run. Do not use for promotion.",
4 "parameters": {
5 "type": "object",
6 "properties": {
7 "run_id": {"type": "string"},
8 "include_failures": {"type": "boolean"},
9 },
10 "required": ["run_id"],
11 "additionalProperties": False,
12 },
13}
14
15parameters = TOOL["parameters"]
16print(f"tool_name: {TOOL['name']}")
17print(f"required: {parameters['required']}")
18print(f"accepts_extra_fields: {parameters['additionalProperties']}")1tool_name: get_eval_run
2required: ['run_id']
3accepts_extra_fields: FalseA 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 caller owns the project or that a write action is permitted.
The model's output crosses a trust boundary. Even when an API offers constrained or strict structured output, 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.
1class CallRejected(ValueError):
2 pass
3
4def validate_eval_call(call: dict[str, object]) -> dict[str, object]:
5 if call.get("name") != "get_eval_run":
6 raise CallRejected("unknown tool")
7 args = call.get("args")
8 if not isinstance(args, dict):
9 raise CallRejected("args must be an object")
10 allowed = {"run_id", "include_failures"}
11 unknown = set(args) - allowed
12 if unknown:
13 raise CallRejected(f"unknown fields: {sorted(unknown)}")
14 if not isinstance(args.get("run_id"), str):
15 raise CallRejected("run_id must be a string")
16 if "include_failures" in args and not isinstance(args["include_failures"], bool):
17 raise CallRejected("include_failures must be a boolean")
18 return args
19
20candidates = [
21 {"name": "get_eval_run", "args": {"run_id": "R42", "include_failures": True}},
22 {"name": "get_eval_run", "args": {"run_id": "R42", "promote_now": True}},
23]
24for call in candidates:
25 try:
26 args = validate_eval_call(call)
27 print(f"accepted: {args['run_id']}")
28 except CallRejected as exc:
29 print(f"rejected: {exc}")1accepted: R42
2rejected: unknown fields: ['promote_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.
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.
1import json
2
3EVAL_RUNS = {
4 "R42": {"status": "failed", "metric": "citation_precision", "score": "0.81"},
5}
6
7class ScriptedModel:
8 def __init__(self) -> None:
9 self.turns = 0
10
11 def respond(self, messages: list[dict[str, object]]) -> dict[str, object]:
12 self.turns += 1
13 observations = [item for item in messages if item["role"] == "tool"]
14 if not observations:
15 return {
16 "role": "assistant",
17 "tool_call": {
18 "id": "eval-1",
19 "name": "get_eval_run",
20 "args": {"run_id": "R42"},
21 },
22 }
23 result = json.loads(str(observations[-1]["content"]))
24 return {
25 "role": "assistant",
26 "content": (
27 f"Eval run R42 {result['status']} because {result['metric']} "
28 f"scored {result['score']}."
29 ),
30 }
31
32def execute_eval_tool(call: dict[str, object]) -> dict[str, str]:
33 if call.get("name") != "get_eval_run":
34 raise ValueError("tool not allowed")
35 args = call.get("args")
36 if not isinstance(args, dict) or set(args) != {"run_id"}:
37 raise ValueError("expected only run_id")
38 run_id = args["run_id"]
39 if not isinstance(run_id, str) or run_id not in EVAL_RUNS:
40 raise ValueError("unknown run")
41 return EVAL_RUNS[run_id]
42
43model = ScriptedModel()
44messages: list[dict[str, object]] = [
45 {"role": "user", "content": "Why did eval run R42 fail?"}
46]
47
48first = model.respond(messages)
49call = first["tool_call"]
50messages.append(first)
51observation = execute_eval_tool(call) # runtime executes, not model
52messages.append(
53 {"role": "tool", "tool_call_id": call["id"], "content": json.dumps(observation)}
54)
55final = model.respond(messages)
56
57print(f"requested_tool: {call['name']}")
58print(f"tool_status: {observation['status']}")
59print(f"answer: {final['content']}")
60print(f"model_turns: {model.turns}")1requested_tool: get_eval_run
2tool_status: failed
3answer: Eval run R42 failed because citation_precision scored 0.81.
4model_turns: 2The 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.[2][3] The engineering burden remains in your runtime.
Valid arguments can still request a harmful or unauthorized action. An eval lookup is read-only; promote_model changes production traffic. Writes need more checks:
1from dataclasses import dataclass
2
3@dataclass(frozen=True)
4class Session:
5 project_id: str
6
7MODEL_RUNS = {
8 "R42": {"project_id": "search", "gates_passed": True, "candidate": "reranker-v7"},
9}
10PROMOTIONS: dict[str, str] = {}
11
12def promote_model(session: Session, run_id: str, confirmed: bool) -> str:
13 run = MODEL_RUNS.get(run_id)
14 if run is None:
15 return "blocked: unknown run"
16 if session.project_id != run["project_id"]:
17 return "blocked: project ownership failed"
18 if not run["gates_passed"]:
19 return "blocked: release policy failed"
20 if not confirmed:
21 return "blocked: confirmation required"
22 key = f"promotion:{run_id}"
23 if key in PROMOTIONS:
24 return f"replayed: promotion already exists for {PROMOTIONS[key]}"
25 PROMOTIONS[key] = run["candidate"]
26 return f"promoted: {PROMOTIONS[key]}"
27
28print(promote_model(Session("search"), "Z99999", confirmed=True))
29print(promote_model(Session("ads"), "R42", confirmed=True))
30print(promote_model(Session("search"), "R42", confirmed=False))
31print(promote_model(Session("search"), "R42", confirmed=True))
32print(promote_model(Session("search"), "R42", confirmed=True))1blocked: unknown run
2blocked: project ownership failed
3blocked: confirmation required
4promoted: reranker-v7
5replayed: promotion already exists for reranker-v7Schema 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.
Validation narrows what the model can ask for, but how the runtime uses those arguments matters just as much. When a tool touches a database, a shell, or any other interpreter, pass the validated arguments as bound parameters, never as interpolated strings. A run_id that cleared your schema check still becomes an injection vector the moment you build f"SELECT * FROM runs WHERE id = '{run_id}'" or hand it to a shell with subprocess.run(cmd, shell=True). Use the driver's placeholder binding, cursor.execute("SELECT * FROM runs WHERE id = %s", (run_id,)), and pass process arguments as an argv list with shell=False. The model proposes values; parameterized execution keeps those values as data instead of letting them turn into code.
Tool calls fail in ordinary ways: the model uses an old field name, a run 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.
1EVAL_RUNS = {"R42": {"status": "failed"}}
2
3def execute(call: dict[str, object]) -> dict[str, object]:
4 if call.get("name") != "get_eval_run":
5 return {"ok": False, "error": "tool not allowed"}
6 args = call.get("args")
7 if not isinstance(args, dict):
8 return {"ok": False, "error": "args must be an object"}
9 unknown = sorted(set(args) - {"run_id"})
10 if unknown:
11 return {"ok": False, "error": f"unknown fields: {unknown}"}
12 if "run_id" not in args:
13 return {"ok": False, "error": "required field: run_id"}
14 run_id = args["run_id"]
15 if not isinstance(run_id, str) or run_id not in EVAL_RUNS:
16 return {"ok": False, "error": "unknown run"}
17 return {"ok": True, "status": EVAL_RUNS[run_id]["status"]}
18
19model_attempts = [
20 {"name": "get_eval_run", "args": {}},
21 {"name": "get_eval_run", "args": {"run_id": "R42"}},
22]
23for turn, call in enumerate(model_attempts, start=1):
24 observation = execute(call)
25 print(f"turn_{turn}: {observation}")
26 if observation["ok"]:
27 break1turn_1: {'ok': False, 'error': 'required field: run_id'}
2turn_2: {'ok': True, 'status': 'failed'}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.
1import json
2
3attempts = [
4 {"name": "get_eval_run", "args": {"eval_id": "R42"}},
5 {"name": "get_eval_run", "args": {"eval_id": "R42"}},
6 {"name": "get_eval_run", "args": {"run_id": "R42"}},
7]
8seen: set[str] = set()
9max_turns = 3
10
11def rejection_for(call: dict[str, object]) -> str | None:
12 if call.get("name") != "get_eval_run":
13 return "tool not allowed"
14 args = call.get("args")
15 if not isinstance(args, dict):
16 return "args must be an object"
17 unknown = sorted(set(args) - {"run_id"})
18 if unknown:
19 return f"unknown fields: {unknown}"
20 if "run_id" not in args:
21 return "required field: run_id"
22 return None
23
24for turn, call in enumerate(attempts[:max_turns], start=1):
25 error = rejection_for(call)
26 if error is None:
27 print(f"turn_{turn}: accepted")
28 break
29 fingerprint = json.dumps(call, sort_keys=True)
30 if fingerprint in seen:
31 print(f"turn_{turn}: stopped repeated rejected call")
32 break
33 seen.add(fingerprint)
34 print(f"turn_{turn}: rejected: {error}")1turn_1: rejected: unknown fields: ['eval_id']
2turn_2: stopped repeated rejected callTrack 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.
Alex asks, "Compare eval runs R42 and R43, then promote the safer one if it passes." Those two eval reads are independent. Your runtime may execute them concurrently after validating both calls. A request to promote one candidate based on those results can't run at the same time: the write depends on what the reads discover.
1import asyncio
2
3STATUSES = {
4 "R42": "failed",
5 "R43": "passed",
6}
7
8async def get_eval_status(run_id: str) -> tuple[str, str]:
9 await asyncio.sleep(0)
10 return run_id, STATUSES[run_id]
11
12async def main() -> None:
13 run_ids = ["R42", "R43"]
14 rows = await asyncio.gather(*(get_eval_status(run_id) for run_id in run_ids))
15 for run_id, status in sorted(rows):
16 print(f"{run_id}: {status}")
17 print("write_action: wait for validated read results")
18
19asyncio.run(main())1R42: failed
2R43: passed
3write_action: wait for validated read resultsConcurrency saves elapsed time only when actions are independent. For writes on the same model target, serialize execution and apply idempotency rules.
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.[4]
1import re
2from dataclasses import dataclass
3
4@dataclass(frozen=True)
5class Tool:
6 name: str
7 description: str
8
9catalog = [
10 Tool("get_eval_run", "eval run metric failure score status"),
11 Tool("promote_model", "promote model release production traffic"),
12 Tool("search_eval_policy", "find release policy gate thresholds"),
13]
14allowed = {"get_eval_run", "search_eval_policy"}
15query = "Why did eval run R42 fail its metric?"
16query_terms = set(re.findall(r"[a-z_]+", query.lower()))
17
18ranked = sorted(
19 (
20 (len(query_terms & set(tool.description.split())), tool.name)
21 for tool in catalog
22 if tool.name in allowed
23 ),
24 reverse=True,
25)
26visible_tools = [name for score, name in ranked if score > 0]
27
28print(f"model_visible_tools: {visible_tools}")
29print(f"promotion_tool_exposed: {'promote_model' in visible_tools}")1model_visible_tools: ['get_eval_run']
2promotion_tool_exposed: FalseRouting isn't authorization. The allowlist is applied first. A highly relevant but forbidden write tool must stay unavailable.
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_eval_run for a live eval-status question |
| Arguments | Uses allowed keys and exact run 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.[5][6] Your own release gate needs the schemas, policy failures, and side-effect boundaries your users will hit.
1def score(events: list[dict[str, object]]) -> tuple[bool, str]:
2 if [event.get("type") for event in events] != ["call", "observation", "answer"]:
3 return False, "wrong call sequence"
4
5 call, observation, answer = events
6 if call.get("name") != "get_eval_run":
7 return False, "wrong call sequence"
8 if call.get("args") != {"run_id": "R42"}:
9 return False, "wrong arguments"
10 if not call.get("id") or observation.get("call_id") != call["id"]:
11 return False, "observation mismatched call"
12 if observation.get("run_id") != "R42" or observation.get("status") != "failed":
13 return False, "missing grounded observation"
14 if answer.get("run_id") != observation.get("run_id"):
15 return False, "answer ignored observation"
16 if answer.get("status") != observation.get("status"):
17 return False, "answer ignored observation"
18 return True, "trajectory passed"
19
20good = [
21 {
22 "type": "call",
23 "id": "eval-1",
24 "name": "get_eval_run",
25 "args": {"run_id": "R42"},
26 },
27 {"type": "observation", "call_id": "eval-1", "run_id": "R42", "status": "failed"},
28 {"type": "answer", "run_id": "R42", "status": "failed", "text": "Eval run R42 failed."},
29]
30bad_order = [
31 {"type": "observation", "call_id": "eval-1", "run_id": "R42", "status": "failed"},
32 {"type": "call", "id": "eval-1", "name": "get_eval_run", "args": {"run_id": "R42"}},
33 {"type": "answer", "run_id": "R42", "status": "failed", "text": "Eval run R42 failed."},
34]
35bad_semantics = [
36 {"type": "call", "id": "eval-1", "name": "get_eval_run", "args": {"run_id": "R42"}},
37 {"type": "observation", "call_id": "eval-1", "run_id": "R42", "status": "failed"},
38 {"type": "answer", "run_id": "R42", "status": "passed", "text": "R42 did not fail; an older check was marked failed."},
39]
40print(score(good))
41print(score(bad_order))
42print(score(bad_semantics))1(True, 'trajectory passed')
2(False, 'wrong call sequence')
3(False, 'answer ignored observation')Once correctness is scored, add serving constraints. A runtime that succeeds after ten retries isn't ready for a customer-facing workflow.
1runs = [
2 {"passed": True, "unsafe_writes": 0, "rounds": 2, "latency_ms": 430, "cost_cents": 2.1},
3 {"passed": True, "unsafe_writes": 0, "rounds": 2, "latency_ms": 510, "cost_cents": 2.4},
4 {"passed": False, "unsafe_writes": 0, "rounds": 3, "latency_ms": 680, "cost_cents": 3.8},
5 {"passed": True, "unsafe_writes": 0, "rounds": 2, "latency_ms": 470, "cost_cents": 2.2},
6]
7
8success_rate = sum(run["passed"] for run in runs) / len(runs)
9unsafe_writes = sum(run["unsafe_writes"] for run in runs)
10max_rounds = max(run["rounds"] for run in runs)
11max_latency_ms = max(run["latency_ms"] for run in runs)
12max_cost_cents = max(run["cost_cents"] for run in runs)
13latency_budget_ms = 600
14cost_budget_cents = 3.0
15ready = (
16 success_rate >= 0.75
17 and unsafe_writes == 0
18 and max_rounds <= 3
19 and max_latency_ms <= latency_budget_ms
20 and max_cost_cents <= cost_budget_cents
21)
22
23print(f"success_rate: {success_rate:.0%}")
24print(f"unsafe_writes: {unsafe_writes}")
25print(f"max_rounds: {max_rounds}")
26print(f"max_latency_ms: {max_latency_ms}")
27print(f"latency_budget_ms: {latency_budget_ms}")
28print(f"max_cost_cents: {max_cost_cents:.1f}")
29print(f"cost_budget_cents: {cost_budget_cents:.1f}")
30print(f"release_candidate: {ready}")1success_rate: 75%
2unsafe_writes: 0
3max_rounds: 3
4max_latency_ms: 680
5latency_budget_ms: 600
6max_cost_cents: 3.8
7cost_budget_cents: 3.0
8release_candidate: FalseThe 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 tasks, actual model calls, sandboxed tool results, and labeled policy outcomes before shipping.
The examples 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 doesn't 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.
get_eval_run contract.Extend complete-tool-loop.py with a second read-only tool, get_release_policy(run_id), and a protected write tool, promote_model(run_id). Build five labeled fixtures: two straightforward eval-status questions, one unknown run, one promotion candidate that passes gates, and one candidate that fails gates. Your artifact is a trajectory report containing final outcome, blocked writes, retries, tool rounds, and maximum latency for every fixture.
Answer every question, then check your score. Score above 75% to mark this lesson complete.
9 questions remaining.
Function calling
OpenAI · 2026 · OpenAI API Docs
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
Questions and insights from fellow learners.