Function / tool calling.

Anatomy of a Tool

A tool definition is exactly the schema work from Topic 4, described from the other direction: instead of shaping the model's output, you're shaping the arguments to a function the model can request. Same Pydantic model, new job.

from pydantic import BaseModel, Field

class GetWeatherArgs(BaseModel):
    city: str = Field(description="City name, e.g. 'Bengaluru'")
    unit: str = Field(default="celsius", description="'celsius' or 'fahrenheit'")

def get_weather(city: str, unit: str = "celsius") -> dict:
    """The real implementation — a fake lookup here, a real API call in production."""
    return {"city": city, "temp": 28, "unit": unit, "condition": "clear"}

# OpenAI's SDK can derive the JSON schema straight from the Pydantic model
tool_schema = {
    "type": "function",
    "function": {
        "name": "get_weather",
        "description": "Get the current weather for a city.",
        "parameters": GetWeatherArgs.model_json_schema(),
    },
}

The Tool-Calling Loop

One call is rarely the whole story. The full loop: send messages + tools, check if the model wants to call something, execute it, append the result, call again — until the model responds with text instead of a tool call.

import json
from openai import OpenAI

client = OpenAI()
TOOLS = {"get_weather": get_weather}  # dispatch table: name -> real function

messages = [{"role": "user", "content": "What's the weather in Bengaluru right now?"}]

while True:
    response = client.chat.completions.create(
        model="gpt-4.1-mini",
        messages=messages,
        tools=[tool_schema],
    )
    message = response.choices[0].message
    messages.append(message)  # the model's turn, including any tool_calls

    if not message.tool_calls:
        print(message.content)  # final answer — exit the loop
        break

    for call in message.tool_calls:
        args = json.loads(call.function.arguments)
        result = TOOLS[call.function.name](**args)  # your code actually runs it
        messages.append({
            "role": "tool",
            "tool_call_id": call.id,
            "content": json.dumps(result),
        })

Three Providers, One Loop

The shape is the same everywhere: describe tools, get a request back, execute, send the result. The field names differ.

r = client.chat.completions.create(
    model="gpt-4.1-mini",
    messages=messages,
    tools=[tool_schema],
)
call = r.choices[0].message.tool_calls[0]
name = call.function.name
args = json.loads(
    call.function.arguments)

r = client.models.generate_content(
    model="gemini-2.5-flash",
    contents=contents,
    config={"tools": [{
        "function_declarations":
            [gemini_schema]}]},
)
call = r.candidates[0].content.parts[0]
    .function_call
name, args = call.name, dict(call.args)

r = client.messages.create(
    model="claude-sonnet-4-5",
    max_tokens=300,
    tools=[claude_tool_schema],
    messages=messages,
)
call = next(b for b in r.content
            if b.type == "tool_use")
name, args = call.name, call.input

Multiple Tools & Tool Choice

Real assistants offer several tools at once and let the model pick — or you can force a specific one, the same way Topic 4's Claude example forced a single tool for structured extraction.

Parallel Tool Calls

Modern models can request several tool calls in a single turn — "get the weather in Bengaluru and Mumbai" produces two tool_calls at once instead of two separate round trips. Your loop from Section 1 already handles this: it iterates message.tool_calls, so multiple calls just mean multiple loop iterations before the next model turn.

for call in message.tool_calls:            # could be 1, could be 5
    args = json.loads(call.function.arguments)
    result = TOOLS[call.function.name](**args)
    messages.append({"role": "tool", "tool_call_id": call.id,
                      "content": json.dumps(result)})
# send all results back together in the next request — not one at a time

Errors Inside a Tool

When a tool call fails — bad arguments, a downstream API timeout, a not-found result — don't let the exception crash the loop. Send the error back to the model as the tool result, the same way an API returns a 404 body instead of dropping the connection. The model can often recover: retry with different arguments, apologize, or try a different tool.

for call in message.tool_calls:
    args = json.loads(call.function.arguments)
    try:
        result = TOOLS[call.function.name](**args)
        content = json.dumps(result)
    except Exception as e:
        content = json.dumps({"error": str(e)})  # let the model see and react to this
    messages.append({"role": "tool", "tool_call_id": call.id, "content": content})

Security: Least Privilege

This is where Topic 3's prompt injection section stops being theoretical. Once a model can call tools, injected text doesn't just make it say something wrong — it can make it do something wrong: send an email, delete a record, transfer data. The defense is the same principle every backend engineer already applies to a service account.

Capstone: A Two-Tool Assistant

Build the full loop from Section 1 with two tools, forcing at least one multi-step exchange (Section 4's parallel calls or a sequential follow-up) and proper error handling (Section 5).

import json
from pydantic import BaseModel, Field
from openai import OpenAI

class GetWeatherArgs(BaseModel):
    city: str = Field(description="City name")

class CalculateArgs(BaseModel):
    expression: str = Field(description="A simple arithmetic expression, e.g. '28 - 4'")

def get_weather(city: str) -> dict:
    fake_data = {"Bengaluru": 28, "Mumbai": 32, "Delhi": 24}
    if city not in fake_data:
        raise ValueError(f"no weather data for {city}")
    return {"city": city, "temp_celsius": fake_data[city]}

def calculate(expression: str) -> dict:
    allowed = set("0123456789+-*/(). ")
    if not set(expression) <= allowed:
        raise ValueError("expression contains disallowed characters")
    return {"result": eval(expression)}  # safe here only because of the allowlist above

TOOLS = {"get_weather": get_weather, "calculate": calculate}
TOOL_SCHEMAS = [
    {"type": "function", "function": {"name": "get_weather",
        "description": "Get current temperature for a city.",
        "parameters": GetWeatherArgs.model_json_schema()}},
    {"type": "function", "function": {"name": "calculate",
        "description": "Evaluate a simple arithmetic expression.",
        "parameters": CalculateArgs.model_json_schema()}},
]

def run_assistant(user_message: str) -> str:
    client = OpenAI()
    messages = [{"role": "user", "content": user_message}]

    for _ in range(6):  # guardrail: cap iterations, per Section 5
        response = client.chat.completions.create(
            model="gpt-4.1-mini", messages=messages, tools=TOOL_SCHEMAS,
        )
        message = response.choices[0].message
        messages.append(message)

        if not message.tool_calls:
            return message.content

        for call in message.tool_calls:
            args = json.loads(call.function.arguments)
            try:
                result = TOOLS[call.function.name](**args)
                content = json.dumps(result)
            except Exception as e:
                content = json.dumps({"error": str(e)})
            messages.append({"role": "tool", "tool_call_id": call.id, "content": content})

    return "gave up after too many tool calls"


if __name__ == "__main__":
    print(run_assistant(
        "What's the temperature difference between Bengaluru and Mumbai?"
    ))