Function Calling in Python: A Working Implementation

Function calling looks like three lines of code in the quickstart and turns into a state machine in production. The model does not execute anything; it emits a structured request to execute something, and every part of the reliability story lives in the code you write around that request.

This is the Python version of that code — the loop, the validation, the error paths and the stop condition. The conceptual side is covered in the explanation of how tool calling works; what follows assumes you already accept the premise and want the implementation to hold up.

Describe the tool once, in one place

Hand-writing JSON Schema next to a Python function guarantees they drift. The schema says the argument is called path, the function signature says file_path, and the failure shows up as a confusing runtime TypeError weeks later.

Generate the schema from a typed model instead. Pydantic gives you this directly, and the output is ordinary JSON Schema that any OpenAI-compatible provider accepts.

from pydantic import BaseModel, Field

class ReadFile(BaseModel):
    """Read a UTF-8 text file from the project directory."""
    path: str = Field(description="Path relative to the project root")
    max_bytes: int = Field(default=8192, ge=1, le=1_000_000)

tool = {
    "type": "function",
    "function": {
        "name": "read_file",
        "description": ReadFile.__doc__,
        "parameters": ReadFile.model_json_schema(),
    },
}

The same model then validates what comes back. One definition produces the contract you send and the parser you run on the response, which removes an entire class of mismatch bugs. The wider question of how to shape those schemas so a model uses them correctly is the subject of designing tool schemas the model can actually use.

The loop, in full

The structure is fixed: send messages, inspect the reply, and if it contains tool calls, execute them, append the results, and send again. Termination happens when the model replies with content and no tool calls.

def run(client, model, messages, tools, registry, max_turns=8):
    for _ in range(max_turns):
        resp = client.chat.completions.create(
            model=model, messages=messages, tools=tools
        )
        msg = resp.choices[0].message
        messages.append(msg.model_dump(exclude_none=True))

        if not msg.tool_calls:
            return msg.content

        for call in msg.tool_calls:
            messages.append({
                "role": "tool",
                "tool_call_id": call.id,
                "content": dispatch(registry, call),
            })
    raise RuntimeError("tool loop did not converge")

Two details are load-bearing. The assistant message containing the tool calls must be appended to history before the results, and every result must carry the exact tool_call_id it answers. Providers reject the request outright when a tool message references an id that never appeared, which is one of the more common 400s described in the reference of LLM API error codes.

The turn cap is not decoration. Without it a model that keeps re-calling the same tool will run until your budget notices, which is exactly the pattern discussed in spotting agents that have started looping.

Arguments arrive as a string, and it may not be valid

The arguments field is a JSON string, not an object, and nothing guarantees it parses or matches your schema. Long argument payloads truncated by an output cap are a real source of malformed JSON, and so are models emitting a number where you asked for a string.

def dispatch(registry, call):
    fn, model_cls = registry[call.function.name]
    try:
        args = model_cls.model_validate_json(call.function.arguments)
    except ValidationError as e:
        return json.dumps({"error": "invalid_arguments",
                           "detail": e.errors()})
    try:
        return json.dumps({"ok": True, "result": fn(args)})
    except Exception as e:
        return json.dumps({"error": type(e).__name__, "detail": str(e)})

Notice that both failure paths return a value rather than raising. A validation error handed back to the model as a tool result is information it can act on — it will usually correct the argument and try again. An exception that escapes the loop is a dead conversation.

Return errors, but bound them

Returning errors to the model is powerful and easy to overdo. A stack trace in a tool result burns hundreds of tokens and invites the model to reason about your internals rather than fix its call.

Keep tool error payloads short, structured and actionable: what was wrong, and what a valid value looks like. Truncate any free text to a couple of hundred characters. If the same tool fails twice in a row with the same error, stop returning it and end the turn — the model is not converging and further attempts just cost money, a dynamic examined in the hidden cost of retries.

Parallel calls and the concurrency question

Models routinely emit several tool calls in one assistant message. Executing them sequentially is correct but often slow, and the calls are usually independent reads.

A thread pool is enough for I/O-bound tools. The constraint is ordering: results may complete in any order, but you must append them to the message list in the order the calls appeared, because some providers validate that correspondence. Collect into a dict keyed by call id, then emit in the original sequence.

Be careful about which tools are safe to run in parallel. Two reads are fine; a write and a read of the same resource are not. If any call in the batch mutates state, run that batch sequentially rather than reasoning about interleavings you did not design.

Streaming makes assembly your job

When you stream a response that contains tool calls, the arguments arrive as fragments across many chunks. Each delta carries an index identifying which call it belongs to, and you accumulate the string yourself.

calls = {}
for chunk in stream:
    for d in chunk.choices[0].delta.tool_calls or []:
        slot = calls.setdefault(d.index, {"id": "", "name": "", "args": ""})
        if d.id: slot["id"] = d.id
        if d.function.name: slot["name"] = d.function.name
        if d.function.arguments: slot["args"] += d.function.arguments

Only parse once the stream has finished cleanly. Parsing a partial argument string produces a confusing error that looks like a model failure and is actually a transport one — see the notes on detecting truncated SSE streams for how to tell those apart.

What to check before you ship

Run the loop against deliberately hostile inputs: a tool name the registry does not contain, arguments that fail validation, a tool that raises, and a prompt that makes the model call the same tool repeatedly. All four should produce a bounded, logged outcome rather than an exception at the top level.

Then log every call with its name, argument size, duration and outcome. Tool-level telemetry is the difference between "the agent is flaky" and "the search tool times out on queries longer than 80 characters", and it costs one wrapper function to add.