Step 1: One round trip: what the model actually sends back
An agent is a loop around one API call. Before writing the loop, look at what that call returns when tools are attached, because everything the loop does is a reaction to that shape.
tools.py holds six plain Python functions for Northwind's operations
team: lookup_order, list_orders, convert_units, calculator,
shipping_quote and refund. Next to them, TOOL_SPECS describes the same
functions to the model in the chat completions tool format: a name, a
description, and a JSON schema for the parameters. The model never runs
anything. It reads the descriptions, and when a question needs a tool it
returns an assistant message whose content is empty and whose
tool_calls list says which function to call and with what arguments.
Two details of that message decide how the rest of the lab is written. Each
tool call carries an id, and the API will refuse the next request unless
every id gets a matching tool message back. And function.arguments is a
string containing JSON rather than a parsed object. The model produced it
token by token, and on a bad day it is not valid JSON at all. Parsing it is
the executor's job, later; extracting it is this step's.
do this
1. Run. agent.py --one-shot sends one completion with the tools and
prints the raw assistant message: the finish reason is tool_calls, the
content is None, and there is a call to lookup_order with the order id
in its arguments string. Below it, parsed: is empty. That is the gap you
close now.
2. Implement parse_tool_calls(message) in agent.py. Walk
message.tool_calls (it is None on a plain text answer, so guard for
that) and collect one dict per call: {"id": tc.id, "name": tc.function.name, "arguments": tc.function.arguments}. Keep arguments
as the raw string. Return [] when there are no tool calls.
3. Run again and read oneshot.json. The parsed list now mirrors the
raw calls. Note what the model did with the question: it asked for the
order first, because the shipping quote needs a weight and a country it
does not have yet. It cannot get them in this step, since nothing runs the
tool and nothing sends the result back. That is the loop, and it is Step 3.
Why this matters in practice
Every framework's agent loop starts from this exact message. Knowing that the arguments are a string, that ids must be answered, and that a text answer means "no more tool calls" is enough to read any of them.
agent.py, the file you edit250 lines
"""agent.py: a tool-using agent on the chat completions API and nothing else.
python3 agent.py "How much would it cost to ship order NW-1007 to its destination?"
python3 agent.py --one-shot "..." # one completion, print the raw assistant message
python3 agent.py --questions q.jsonl --out runs.jsonl
python3 agent.py --scripted loop # replay a recorded transcript offline (see fake_client.py)
Options: --model, --max-steps, --max-tool-calls, --token-budget, --max-result-chars,
--trace trace.jsonl, --auto-approve, --audit audit.jsonl
"""
import argparse
import json
import os
import sys
import time
import tools
from tools import TOOL_SPECS, SIDE_EFFECT_TOOLS
MODEL = os.environ.get("AGENT_MODEL", "meta/llama-3.3-70b-instruct")
BASE_URL = os.environ.get("AGENT_BASE_URL", "http://nim-proxy.labs.svc:8080/v1")
APPROVAL_LIMIT_USD = float(os.environ.get("AGENT_APPROVAL_LIMIT", "100"))
SYSTEM = (
"You are Northwind Cloud's operations assistant. You have tools for order lookup, "
"arithmetic, unit conversion, shipping quotes and refunds. Use a tool for every lookup, "
"calculation, conversion and quote, including ones that look trivial: call calculator for "
"any arithmetic and convert_units for any unit conversion; never compute or convert numbers "
"yourself. "
"When the tools have given you what you need, answer in one or two sentences with the "
"final numbers. If a tool reports an error, say so plainly instead of inventing a value."
)
def make_client():
from openai import OpenAI
return OpenAI(base_url=BASE_URL, api_key="not-needed")
# ----------------------------------------------------------------- Step 1
def parse_tool_calls(message):
"""Extract the tool calls from an assistant message.
Returns a list of {"id", "name", "arguments"} where arguments is the RAW
JSON string the model produced (parsing it is the executor's job).
An assistant message with no tool calls returns [].
"""
# TODO (Step 01): walk message.tool_calls (it may be None) and collect
# {"id": tc.id, "name": tc.function.name, "arguments": tc.function.arguments}
# arguments stays the RAW string; return [] when there are no tool calls.
return []
# ----------------------------------------------------------------- Step 2
def build_registry(specs, module=tools):
"""Map every tool name declared in specs to the Python function of that
name in `module`. A declared tool with no implementation is a bug: raise."""
return {}
def dispatch(registry, name, args):
"""Run one tool and return the JSON string that goes back to the model.
An unknown tool name returns a JSON error rather than raising."""
return json.dumps({"error": "dispatch arrives in Step 02"})
# ----------------------------------------------------------------- Step 6
def check_policy(agent, name, args):
"""Policy hook, called before any tool runs. Return None to allow the
call, or a string explaining the denial; the string goes back to the
model as the tool result. Every decision on a side-effecting tool is
written to the audit log."""
return None # the policy hook arrives in Step 06
def run_tool(agent, name, args):
"""Policy first, then dispatch. Returns the tool message content."""
denial = check_policy(agent, name, args)
if denial:
return json.dumps({"error": denial, "executed": False})
return dispatch(agent.registry, name, args)
# ----------------------------------------------------------------- Step 5
def execute_call(agent, call):
"""Turn one parsed tool call into the content of its tool message.
Never raises: arguments that are not a JSON object, an unknown tool,
wrong parameter names and a tool that throws all become
{"error": "..."} so the model can read what went wrong.
"""
return run_tool(agent, call["name"], json.loads(call["arguments"]))
class Result:
def __init__(self, question):
self.question = question
self.answer = None
self.steps = 0 # completions requested
self.tool_calls = [] # [{"name", "arguments", "result"}]
self.tokens = 0
self.stopped_by = None # None, or "max_steps" / "max_tool_calls" / "token_budget" / "provider_error"
self.duration_ms = 0.0
self.messages = []
def to_dict(self):
return {"question": self.question, "answer": self.answer, "steps": self.steps,
"tool_calls": self.tool_calls, "tokens": self.tokens, "stopped_by": self.stopped_by,
"duration_ms": round(self.duration_ms, 1), "messages": self.messages}
class Agent:
def __init__(self, client, model=MODEL, specs=TOOL_SPECS, registry=None, max_steps=8,
max_tool_calls=20, token_budget=30000, max_result_chars=1500,
auto_approve=False, approval_limit=APPROVAL_LIMIT_USD, trace_path=None,
audit_path="audit.jsonl", retries=1, temperature=0):
self.client = client
self.model = model
self.specs = specs
self.registry = registry if registry is not None else build_registry(specs)
self.max_steps = max_steps
self.max_tool_calls = max_tool_calls
self.token_budget = token_budget
self.max_result_chars = max_result_chars
self.auto_approve = auto_approve
self.approval_limit = approval_limit
self.trace_path = trace_path
self.audit_path = audit_path
self.retries = retries
self.temperature = temperature
# ------------------------------------------------------------- Step 3
def run(self, question):
"""The loop. Send the conversation with the tools, read the tool calls,
run them, append one tool message per call, repeat until the model
answers in text. Three hooks are called from inside the loop and are
filled in by later steps: over_budget (Step 4), call_model (Step 5),
on_tool_result (Step 7)."""
raise NotImplementedError("the loop arrives in Step 03")
# ------------------------------------------------------------- Step 4
def over_budget(self, result):
"""Return the name of the exhausted budget, or None. Checked before
every completion, so a runaway model is cut off at the limit."""
return None # budgets arrive in Step 04
# ------------------------------------------------------------- Step 5
def call_model(self, messages):
"""One completion request with the tools attached. Returns the response,
or None when the API failed on every attempt (an exception, or a
response with no choices, which gateways return for provider errors)."""
return self.client.chat.completions.create(model=self.model, messages=messages, tools=self.specs,
temperature=self.temperature, max_tokens=400)
# ------------------------------------------------------------- Step 7
def on_tool_result(self, result, call, content, elapsed_ms):
"""Runs on every tool result before it is appended to the conversation.
Cuts oversized results down to max_result_chars with a marker the
model can read, and writes one trace record per call."""
return content # truncation and tracing arrive in Step 07
# ------------------------------------------------------------- given
def trace(self, record):
if self.trace_path:
with open(self.trace_path, "a", encoding="utf-8") as f:
f.write(json.dumps(record) + "\n")
def audit(self, record):
record = {"ts": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()), **record}
with open(self.audit_path, "a", encoding="utf-8") as f:
f.write(json.dumps(record) + "\n")
def log(self, text):
print(f"[agent] {text}", file=sys.stderr)
# ----------------------------------------------------------------- CLI
def one_shot(client, model, question):
"""Send ONE completion with the tools and print the raw assistant message."""
messages = [{"role": "system", "content": SYSTEM}, {"role": "user", "content": question}]
r = client.chat.completions.create(model=model, messages=messages, tools=TOOL_SPECS, temperature=0, max_tokens=300)
m = r.choices[0].message
print("finish_reason:", r.choices[0].finish_reason)
print("content:", repr(m.content))
print("tool_calls (raw):")
for tc in (m.tool_calls or []):
print(f" id={tc.id} name={tc.function.name} arguments={tc.function.arguments!r}")
calls = parse_tool_calls(m)
print("parsed:", json.dumps(calls, indent=2))
with open("oneshot.json", "w", encoding="utf-8") as f:
json.dump({"question": question, "finish_reason": r.choices[0].finish_reason,
"content": m.content, "parsed": calls,
"raw_tool_calls": [{"id": tc.id, "name": tc.function.name, "arguments": tc.function.arguments} for tc in (m.tool_calls or [])]}, f, indent=2)
return calls
def print_result(res):
print(f"Q: {res.question}")
for c in res.tool_calls:
print(f" -> {c['name']}({c['arguments']}) = {c['result'][:110]}")
print(f" steps={res.steps} tool_calls={len(res.tool_calls)} tokens={res.tokens} stopped_by={res.stopped_by} {res.duration_ms:.0f} ms")
print(f"A: {res.answer}\n")
def main(argv):
p = argparse.ArgumentParser()
p.add_argument("question", nargs="?")
p.add_argument("--one-shot", metavar="QUESTION")
p.add_argument("--questions", metavar="FILE.jsonl")
p.add_argument("--out", default="runs.jsonl")
p.add_argument("--scripted", metavar="NAME", help="replay a script from fake_client.SCRIPTS")
p.add_argument("--model", default=MODEL)
p.add_argument("--max-steps", type=int, default=8)
p.add_argument("--max-tool-calls", type=int, default=20)
p.add_argument("--token-budget", type=int, default=30000)
p.add_argument("--max-result-chars", type=int, default=1500)
p.add_argument("--trace", metavar="FILE.jsonl")
p.add_argument("--audit", default="audit.jsonl")
p.add_argument("--auto-approve", action="store_true")
a = p.parse_args(argv)
if a.scripted:
from fake_client import SCRIPTS, ScriptedClient
script = SCRIPTS[a.scripted]
client = ScriptedClient(script["turns"])
questions = [script["question"]]
else:
client = make_client()
if a.one_shot:
one_shot(client, a.model, a.one_shot)
return
if a.questions:
questions = [json.loads(l)["question"] for l in open(a.questions, encoding="utf-8") if l.strip()]
elif a.question:
questions = [a.question]
else:
p.error("give a question, --questions FILE, --one-shot QUESTION or --scripted NAME")
agent = Agent(client, model=a.model, max_steps=a.max_steps, max_tool_calls=a.max_tool_calls,
token_budget=a.token_budget, max_result_chars=a.max_result_chars,
auto_approve=a.auto_approve, trace_path=a.trace, audit_path=a.audit)
with open(a.out, "w", encoding="utf-8") as out:
for q in questions:
res = agent.run(q)
print_result(res)
out.write(json.dumps(res.to_dict()) + "\n")
if __name__ == "__main__":
main(sys.argv[1:])provided for you:fake_client.pyorders.jsonquestions.jsonltools.py
