Why native function calling

In the last topic the model emitted Action: search: … as text and you parsed it with a regex. That works, but it's fragile: the model can format the line slightly wrong, wrap it in prose, or mangle a multi-argument call. Native function calling fixes this — you give the API a machine-readable schema for each tool, and the model returns a structured tool call (validated JSON arguments) in a dedicated field, not free text you have to scrape. Every major provider (OpenAI, Anthropic, Gemini, Mistral) supports it; the shapes differ slightly but the mental model is identical.

Describing a tool: the JSON schema

You declare each tool as a name, a description, and a JSON Schema for its parameters. The description and field docs are the prompt — they're how the model knows when and how to call it.

tools = [{
    "type": "function",
    "function": {
        "name": "get_weather",                      # the tool name the model will call
        "description": "Get today's weather for a city.",   # WHEN to use it — this drives tool selection
        "parameters": {                             # JSON Schema describing the arguments
            "type": "object",
            "properties": {
                "city": {"type": "string", "description": "City name, e.g. 'Paris'"},   # each field is documented
                "unit": {"type": "string", "enum": ["c", "f"], "description": "Temperature unit"},  # enum = allowed values
            },
            "required": ["city"],                   # `city` is mandatory; `unit` is optional
        },
    },
}]
# The model reads this schema and produces arguments that CONFORM to it — no text parsing on your side.

The structured tool-call loop

The loop is the same reason→act→observe shape as before, but the plumbing is structured. The model's reply either contains tool_calls (it wants to act) or plain content (it's done). You execute each call and append a tool-role message carrying the result, keyed by the call's id:

import json

def run(question: str, max_steps: int = 5) -> str:
    messages = [{"role": "user", "content": question}]

    for _ in range(max_steps):                                  # bounded loop (same discipline as Topic 1)
        resp = client.chat.completions.create(
            model="gpt-4o-mini", messages=messages, tools=tools,   # pass the tool schemas every turn
        ).choices[0].message
        messages.append(resp)                                    # record the assistant turn (may hold tool_calls)

        if not resp.tool_calls:                                  # no tool call → the model gave a final answer
            return resp.content

        for call in resp.tool_calls:                             # the model may request SEVERAL calls at once
            name = call.function.name                            # which tool
            args = json.loads(call.function.arguments)           # arguments arrive as a JSON string → parse to dict
            result = TOOLS[name](**args)                         # run it, spreading the dict into kwargs (like fn(...args))
            messages.append({                                    # feed the result back as a TOOL message…
                "role": "tool",
                "tool_call_id": call.id,                         # …tied to THIS call's id so the model matches them up
                "content": json.dumps(result),                  # content is a string; serialize structured results
            })
    return "Stopped: step limit reached."

Compare to Topic 1: no re.search, no format-nudging. The provider guarantees the call is well-formed JSON matching your schema. That reliability is the whole reason to prefer native calling.

Validate arguments — the model can lie

"Well-formed JSON matching the schema" is not "correct". The model can still hallucinate a plausible but wrong value — a city that doesn't exist, a negative price, an id it invented. Schemas enforce shape, not validity. Validate before you act, especially for anything that touches money, data, or external systems:

from pydantic import BaseModel, field_validator     # pydantic = runtime validation (like Zod for Python)

class FlightArgs(BaseModel):                          # declare the expected, VALIDATED shape
    origin: str
    destination: str
    max_price: float

    @field_validator("max_price")                     # a custom rule beyond the type
    @classmethod
    def price_positive(cls, v):
        if v <= 0:
            raise ValueError("max_price must be positive")   # reject nonsense the schema alone would allow
        return v

def search_flights_safe(raw_args: dict):
    try:
        args = FlightArgs(**raw_args)                 # validate + coerce; raises if invalid
    except ValueError as e:
        return {"error": f"invalid arguments: {e}"}   # return an error the model can read and retry from
    return search_flights(args.origin, args.destination, args.max_price)   # only now do the real work

Parallel tool calls

Modern models can request several tool calls in one turn when the calls are independent (e.g. "compare the weather in Paris and Tokyo" → two get_weather calls at once). Notice the loop above already handles this: it iterates for call in resp.tool_calls and appends one tool message per call. Run independent calls concurrently for latency, but append their results in matching tool_call_id order so the model can pair each result to its request.

# Independent calls → run them concurrently, then append each result keyed by its own id.
import concurrent.futures as cf
with cf.ThreadPoolExecutor() as pool:                              # a thread pool (like Promise.all over I/O tasks)
    futures = {pool.submit(TOOLS[c.function.name], **json.loads(c.function.arguments)): c
               for c in resp.tool_calls}                           # kick them all off at once
    for fut in cf.as_completed(futures):                           # as each finishes…
        call = futures[fut]
        messages.append({"role": "tool", "tool_call_id": call.id,  # …record it against its call id
                         "content": json.dumps(fut.result())})

Handling tool errors — the part tutorials skip

Tools call the real world, so they will fail: a timeout, a 404, a rate limit, bad args. The wrong move is to let the exception crash your loop. The right move is to catch it and hand the error back to the model as the observation — a capable agent will then retry, pick another tool, or explain the failure to the user:

def execute(call):
    name = call.function.name
    try:
        args = json.loads(call.function.arguments)   # could raise if the model produced bad JSON
        result = TOOLS[name](**args)                 # could raise: network error, bad value, etc.
        return json.dumps(result)
    except Exception as e:                            # catch EVERYTHING at the tool boundary…
        return json.dumps({"error": str(e)})         # …and return it as data the model can reason about
# The model sees {"error": "..."} as the tool result and can recover, instead of your process dying.

Give errors informative messages ("no flights found under €200" beats "KeyError"), because the model reads them and its recovery is only as good as what you tell it. Combine with a retry cap and the loop bound so a failing tool can't spin forever.

Pitfalls

• Trusting schema-valid args. Schemas guarantee shape, not validity — validate business rules (positive price, real ids) before acting.
• Skipping the tool_call_id. Each tool result must reference its call's id, or the model can't pair results to requests (and multi-call turns break).
• Letting tool exceptions crash the loop. Catch at the tool boundary and return the error as an observation so the agent can recover.
• Uninformative error messages. The model's recovery depends on the error text — make it actionable.
• Weak descriptions. As in Topic 1, the schema's descriptions drive tool selection and argument quality.

Recap

• Native function calling replaces fragile text parsing: you supply JSON tool schemas, the model returns structured, schema-conformant tool calls.
• The tool-call loop is reason→act→observe with structured plumbing: append the assistant turn, execute each tool_call, append a tool-role message keyed by tool_call_id, repeat under a bound.
• Validate arguments (pydantic) for business validity — schema shape isn't correctness; treat model args as untrusted input.
• Support parallel calls (one result per id) and, above all, catch tool errors and return them as observations so the agent recovers.
• Worked examples map to the exercises: a tool schema (Ex 1), the structured loop (Ex 2), argument validation (Ex 3), and error recovery (Ex 4).

tools = [{
    "type": "function",
    "function": {
        "name": "search_flights",
        "description": "Search available flights between two cities under a price cap.",
        "parameters": {
            "type": "object",
            "properties": {
                "origin": {"type": "string", "description": "Departure city"},
                "destination": {"type": "string", "description": "Arrival city"},
                "max_price": {"type": "number", "description": "Maximum price in EUR"},
            },
            "required": ["origin", "destination", "max_price"],
        },
    },
}]

resp = client.chat.completions.create(model="gpt-4o-mini", tools=tools,
    messages=[{"role":"user","content":"Find flights from Paris to Tokyo under €800"}]).choices[0].message
for call in resp.tool_calls:
    print(call.function.name, call.function.arguments)   # search_flights {"origin":"Paris","destination":"Tokyo","max_price":800}

import json
TOOLS = {"search_flights": search_flights, "get_weather": get_weather}

def run(question, max_steps=5):
    msgs = [{"role":"user","content":question}]
    for _ in range(max_steps):
        resp = client.chat.completions.create(model="gpt-4o-mini", messages=msgs, tools=tools
               ).choices[0].message
        msgs.append(resp)
        if not resp.tool_calls:
            return resp.content
        for call in resp.tool_calls:
            args = json.loads(call.function.arguments)
            result = TOOLS[call.function.name](**args)
            msgs.append({"role":"tool","tool_call_id":call.id,"content":json.dumps(result)})
    return "step limit reached"

from pydantic import BaseModel, field_validator

class FlightArgs(BaseModel):
    origin: str
    destination: str
    max_price: float
    @field_validator("max_price")
    @classmethod
    def positive(cls, v):
        if v <= 0: raise ValueError("max_price must be positive")
        return v

def search_flights_safe(raw):
    try:
        a = FlightArgs(**raw)
    except ValueError as e:
        return {"error": f"invalid arguments: {e}"}   # goes back to the model as the observation
    return search_flights(a.origin, a.destination, a.max_price)

def execute(call):
    try:
        args = json.loads(call.function.arguments)
        result = TOOLS[call.function.name](**args)
        if not result:
            return json.dumps({"error": "No flights found under that price."})
        return json.dumps(result)
    except Exception as e:
        return json.dumps({"error": str(e)})