Lesson 3: Tool Execution

Generating tool schemas from signatures

Tool calling inverts the flow. Instead of the model returning data for you to use, it asks you to run a function and tell it the result.

You describe available tools, the model requests one with arguments, your code executes it, and the result goes back into the conversation. The part most often done badly is not the describing. It is everything after the model asks.

Start with the description. Writing tool schemas by hand duplicates information that already exists in your function signatures, and the two drift.

from pydantic import BaseModel, Field


class SearchRecipesArgs(BaseModel):
    """Search the indexed recipe corpus by keyword."""

    query: str = Field(description="Search terms, such as ingredients or a dish name")
    max_results: int = Field(default=5, ge=1, le=20)
    cuisine: str | None = Field(default=None, description="Optional cuisine filter")


def search_recipes(args: SearchRecipesArgs) -> list[dict]:
    """Return recipes matching the query."""
    return index.search(args.query, limit=args.max_results, cuisine=args.cuisine)

The schema for the model comes from the same class that validates the arguments:

def tool_definition(name: str, args_model: type[BaseModel]) -> dict:
    """Build a provider tool definition from a Pydantic argument model."""
    return {
        "name": name,
        "description": (args_model.__doc__ or "").strip(),
        "input_schema": args_model.model_json_schema(),
    }

[VOLATILE: the key names in tool definitions differ by provider, for instance input_schema versus parameters. Verify against the provider you target.]

One definition, two consumers. Change a field and both the description sent to the model and the validation performed on the result change together.

Writing tool descriptions well matters more than the mechanism. The description is a prompt. It is how the model decides whether this tool applies, and vague descriptions produce tools called at the wrong time or not at all. Say what the tool does, when to use it, and what it returns. Name parameters as you would name function parameters, and describe any that are not self-evident.

Keep the tool count moderate. Model accuracy in choosing among tools degrades as the list grows, and thirty tools in one request is both expensive in tokens and unreliable in selection. If you have many, group them behind fewer tools with a mode parameter, or select a relevant subset per request.

The dispatch registry

When the model asks for search_recipes, something must map that string to a callable. The obvious implementations are dangerous.

# Never do this
result = eval(f"{tool_name}(**arguments)")

# Also never
result = globals()[tool_name](**arguments)

Both let a name chosen by the model reach arbitrary code. The model is not adversarial, but its input may be, and Lesson 5 covers exactly how a document can influence what the model asks for. Only tools you explicitly registered should be reachable.

from collections.abc import Callable
from dataclasses import dataclass
from typing import Any


@dataclass(frozen=True)
class RegisteredTool:
    name: str
    args_model: type[BaseModel]
    handler: Callable[[BaseModel], Any]
    timeout_seconds: float = 10.0
    requires_approval: bool = False


class ToolRegistry:
    def __init__(self) -> None:
        self._tools: dict[str, RegisteredTool] = {}

    def register(self, tool: RegisteredTool) -> None:
        if tool.name in self._tools:
            raise ValueError(f"tool {tool.name!r} is already registered")
        self._tools[tool.name] = tool

    def get(self, name: str) -> RegisteredTool | None:
        return self._tools.get(name)

    def definitions(self) -> list[dict]:
        return [tool_definition(t.name, t.args_model) for t in self._tools.values()]

An explicit dictionary, populated deliberately. A name that is not in it returns None, which becomes an error message to the model rather than an exception or, far worse, a call to something you never intended to expose.

The requires_approval and timeout_seconds fields are declared per tool rather than globally, because the right values differ. Searching an index and deleting a file need different treatment, and putting that in the registry means the policy lives next to the tool rather than in the calling code.

Validating arguments before execution

The model is an untrusted caller. It may send a string where an integer belongs, omit a required field, invent a parameter, or supply a value that is well-typed and dangerous.

def execute_tool(registry: ToolRegistry, name: str, raw_args: dict) -> ToolResult:
    """Validate and execute a tool call. Never raises for model mistakes."""
    tool = registry.get(name)
    if tool is None:
        return ToolResult.error(f"Unknown tool {name!r}. Available: {registry.names()}")

    try:
        args = tool.args_model.model_validate(raw_args)
    except ValidationError as exc:
        return ToolResult.error(f"Invalid arguments for {name}: {exc}")

    try:
        with time_limit(tool.timeout_seconds):
            value = tool.handler(args)
    except TimeoutError:
        return ToolResult.error(f"{name} timed out after {tool.timeout_seconds}s")
    except Exception as exc:
        logger.exception("tool %s failed", name)
        return ToolResult.error(f"{name} failed: {type(exc).__name__}")

    return ToolResult.ok(value)

Validation happens before the handler runs, so the handler receives a typed, validated object and contains no defensive checks. This is the boundary principle from Module 2 again, with the model on the untrusted side.

Note the exception handler logs the full detail and returns only the exception type to the model. Internal error messages can contain file paths, query fragments, and configuration details, and everything you return becomes part of the conversation. Log richly, return narrowly.

Returning errors to the model instead of raising

This is the design decision that most distinguishes working tool loops from broken ones.

When a tool call fails, the instinct is to raise. In a tool loop that is usually wrong, because the model can often fix the problem if told what it was.

# The model asked for search_recipes(query="carbonara", limit=5)
# but the parameter is called max_results

# Raising: the whole turn fails, the user sees an error
raise ValidationError(...)

# Returning: the model sees the message and corrects itself
ToolResult.error(
    "Invalid arguments for search_recipes: unexpected field 'limit'. "
    "Did you mean 'max_results'?"
)

The second form usually results in a correct call on the next iteration. This is the repair loop from Lesson 2 in a different setting: give the model the specific error and let it fix its own mistake.

Which failures go back to the model, and which do not. Return argument validation failures, unknown tool names, tool timeouts, not-found results, and business rule violations such as a permission denial, since all of these are things the model can respond to sensibly. Raise instead for failures in your own infrastructure, such as your database being unreachable, and for anything indicating the loop itself is broken. The test is whether a different action by the model could plausibly succeed.

Write error messages the model can act on. "Invalid input" is useless. "The date must be in YYYY-MM-DD format, received '3rd March'" produces a corrected call.

The agent loop

Everything so far assembles into the loop.

MAX_ITERATIONS = 10


def run_agent(user_message: str, registry: ToolRegistry) -> AgentResult:
    """Run the call, execute, feed back cycle until completion or a guard trips."""
    messages = [{"role": "user", "content": user_message}]
    guard = LoopGuard(max_iterations=MAX_ITERATIONS)

    while True:
        stop = guard.check()
        if stop is not None:
            return AgentResult.stopped(messages, reason=stop)

        response = call_provider(messages, tools=registry.definitions())
        messages.append(assistant_message(response))

        tool_calls = extract_tool_calls(response)
        if not tool_calls:
            return AgentResult.completed(messages, text=response_text(response))

        results = []
        for call in tool_calls:
            guard.record_call(call.name, call.arguments)
            result = execute_tool(registry, call.name, call.arguments)
            results.append(tool_result_message(call.id, result))

        messages.extend(results)

Read the shape. Call the model. If it returned no tool calls, it is finished and the loop ends. If it did, execute each one, append the results, and go round again. The guard is checked before every model call, not after, so a stopped loop does not spend one more request.

[IMAGE PROMPT M6-2
Purpose: Show the agent loop cycle including where validation and termination guards sit.
Visual type: Cyclic flow diagram with guard checkpoints.
Prompt: A clean educational cyclic diagram with four main stages arranged in a circle, connected by arrows flowing clockwise. Starting at the top, stage 1 is labelled "call model with messages and tool definitions". Clockwise to the right, a decision diamond labelled "tool calls returned?" with a "no" branch exiting right to a terminal box labelled "return final answer". The "yes" branch continues clockwise to stage 2, a box labelled "validate arguments" with a small side branch to a box labelled "invalid: return error message to model". Continuing clockwise to stage 3, a box labelled "execute handler with timeout". Continuing to stage 4 at the left, a box labelled "append tool results to messages", with an arrow closing the circle back to stage 1. On the arrow entering stage 1, a gate symbol is labelled "loop guard: max iterations, repeated call, no progress", with a branch exiting to a terminal box labelled "stop and report reason".
Required elements: Four numbered stages in a clockwise cycle, the tool calls decision with its exit, the validation side branch feeding back as an error message, the guard gate on the entry to the model call with its own exit, two terminal boxes.
Style: Clean educational illustration, professional, uncluttered, high contrast, flat vector, clear arrowheads showing clockwise direction.
Layout: Circular arrangement with terminals branching outward to the right and lower right.
Text labels: "call model with messages and tool definitions", "tool calls returned?", "return final answer", "validate arguments", "invalid: return error message to model", "execute handler with timeout", "append tool results to messages", "loop guard: max iterations, repeated call, no progress", "stop and report reason", "yes", "no".
Aspect ratio: 4:3
Accessibility: Show flow direction with explicit arrowheads and label every branch in text rather than relying on position or colour.
Avoid: Robot or brain imagery, vendor logos, screenshots, tiny text, watermarks.
Alt text: Cyclic agent loop diagram showing a model call gated by a loop guard, a check for tool calls that exits with a final answer when none are returned, argument validation that returns errors to the model, timed tool execution, and results appended back into the conversation.
END IMAGE PROMPT]

Termination

An unbounded loop is the most common serious bug in agent code. A model that keeps calling tools without converging will run until something stops it, and if nothing does, it runs until your budget is gone.

Three guards, each catching a different failure.

from collections import Counter
from dataclasses import dataclass, field


@dataclass
class LoopGuard:
    max_iterations: int = 10
    max_repeats: int = 3
    max_calls_without_progress: int = 5

    _iterations: int = field(default=0, init=False)
    _call_signatures: Counter = field(default_factory=Counter, init=False)
    _calls_since_new_information: int = field(default=0, init=False)

    def check(self) -> str | None:
        """Return a stop reason, or None to continue."""
        self._iterations += 1
        if self._iterations > self.max_iterations:
            return f"reached the {self.max_iterations} iteration limit"

        repeated = self._call_signatures.most_common(1)
        if repeated and repeated[0][1] > self.max_repeats:
            return f"tool {repeated[0][0][0]!r} called with identical arguments repeatedly"

        if self._calls_since_new_information > self.max_calls_without_progress:
            return "no new information from recent tool calls"

        return None

    def record_call(self, name: str, arguments: dict) -> None:
        signature = (name, json.dumps(arguments, sort_keys=True))
        self._call_signatures[signature] += 1

    def record_result(self, result_hash: str, seen: set[str]) -> None:
        if result_hash in seen:
            self._calls_since_new_information += 1
        else:
            self._calls_since_new_information = 0
            seen.add(result_hash)

Max iterations is the backstop, and it must exist. Ten is a reasonable default for most tasks.

Repeated call detection catches the loop where a model calls the same tool with the same arguments over and over, usually because the result is not what it expected and it tries again identically. Note that the signature includes the arguments, so calling the same tool with different arguments is legitimate progress and is not blocked.

No-progress detection catches the subtler case where the model varies its calls but the results stop being informative, such as searching repeatedly with slightly different phrasings and getting the same empty result.

Stopping is not failing. When a guard trips, return what you have with a clear reason rather than raising. The conversation so far is often useful, and an honest "I could not complete this after ten steps" is a better outcome than a crash, which is the honest degradation principle from Module 5.

Add a cost ceiling alongside the iteration ceiling for anything running unattended. Ten iterations of a cheap call and ten iterations of an expensive one differ by orders of magnitude, and a token budget bounds the thing you actually care about.

Parallel tool calls and result ordering

Models can request several tools in one turn. Executing them concurrently is a real speed gain, and there is one rule that must not be broken.

Every tool result must be matched to its call identifier. Providers give each tool call an id, and the result must carry it back. If results are returned in the wrong order without ids, or ids are mismatched, the model receives one tool's output labelled as another's, and the resulting behaviour is confusing in a way that is very hard to debug.

import asyncio


async def execute_all(registry: ToolRegistry, calls: list[ToolCall]) -> list[dict]:
    """Execute tool calls concurrently, preserving the call id on every result."""
    async def run(call: ToolCall) -> dict:
        result = await execute_tool_async(registry, call.name, call.arguments)
        return tool_result_message(call.id, result)

    return await asyncio.gather(*(run(call) for call in calls))

asyncio.gather returns results in the order the tasks were passed regardless of completion order, and each result carries its own call id, so both mechanisms agree.

Not everything should run in parallel. Tools with side effects that could conflict, such as two writes to the same record, need sequencing. Tools where one depends on another's output cannot be parallel by definition, though a model requesting both in one turn has already decided they are independent. And parallel execution multiplies your concurrent load on whatever those tools call, which is the subject of the next module.

Timeouts on tool execution

Module 5 established that every network call gets a timeout. Tool execution needs its own, at a different level.

A tool that hangs stalls the entire agent loop. The model is waiting, the user is waiting, and no progress is possible. The per-tool timeout in the registry exists for this reason, and the timeout should fit the tool: a search is a few seconds, a code execution might be thirty.

The awkward detail is that timing out a synchronous function in Python is genuinely difficult. Signal-based approaches only work on the main thread on Unix, and threads cannot be forcibly killed. The practical answers are to make tool handlers asynchronous, where asyncio.wait_for works cleanly, to run risky tools in a subprocess that can be terminated, or to ensure every blocking call inside a handler has its own timeout so the handler cannot exceed the sum of them.

Whichever you choose, a tool that times out should return an error message to the model rather than raising, so the loop can continue and the model can try something else.