Part 2: Core Patterns

Tool Use

An agent without tools is a confident liar. It will invent API responses, fabricate database rows, and cite papers that do not exist. All of this happens with the fluency of someone who has done it a thousand times. This chapter gives your agents hands.

6.1 The Hallucination Problem

Consider a customer service agent tasked with checking order status. You have given it a system prompt explaining the database schema, the API endpoints, and the response formats. You ask: "What is the status of order #48291?"

Without tools, here is what happens:

Assistant: Order #48291 was placed on March 3rd and is currently
in transit via FedEx (tracking: 7829104856302). Expected delivery
is March 7th. The shipping address is 142 Oak Lane, Portland, OR.

Every detail in that response — the date, the carrier, the tracking number, the address — is fabricated. The model has never queried your order database. It has no network access. It predicted what a plausible order status response would look like based on its training data, and generated that with absolute confidence. The customer now has a fake tracking number. If they call back angry, a second agent might fabricate a different tracking number.

This is not a rare failure mode. It is the default behavior of any LLM asked to perform an action it cannot actually perform. Language models are completion engines: given a prompt that implies a database lookup happened, they will complete the text as if it did. The more specific and confident the system prompt ("You have access to the order database"), the more specific and confident the hallucination becomes.

Instead of pretending the model can query databases, we give it a structured protocol for requesting that we query databases on its behalf. The model outputs a structured function call. Our code executes it. The real result goes back into the conversation. The model then responds based on actual data. This is tool use, and it is the single most important capability separating toy demos from production agents.

6.2 The Function Calling Protocol

Function calling is a protocol, not a feature of any single model. During text generation, the model can choose to emit a structured tool call instead of (or in addition to) natural language. The host application intercepts this call, executes the corresponding function, and feeds the result back into the conversation as a new message.

Here is the protocol in its simplest form with the OpenAI API:

import openai

response = openai.chat.completions.create(
    model="gpt-4o",
    messages=[
        {"role": "system", "content": "You are an order status assistant."},
        {"role": "user", "content": "What is the status of order #48291?"},
    ],
    tools=[
        {
            "type": "function",
            "function": {
                "name": "get_order_status",
                "description": "Look up the current status of a customer order by its ID.",
                "parameters": {
                    "type": "object",
                    "properties": {
                        "order_id": {
                            "type": "string",
                            "description": "The order ID, e.g. '48291'"
                        }
                    },
                    "required": ["order_id"]
                }
            }
        }
    ],
)

message = response.choices[0].message
print(message.tool_calls)
# [ToolCall(id='call_abc123', function=Function(
#     name='get_order_status', arguments='{"order_id": "48291"}'))]

Several things are happening here that deserve attention:

1. The model does not execute anything. It outputs a JSON object saying "I want to call get_order_status with order_id=48291." Your code decides whether and how to execute that.
2. The tool definition is a JSON Schema. The parameters field follows the JSON Schema specification. The model uses this schema to understand what arguments are valid and how to format them.
3. Each tool call has a unique ID. This ID (call_abc123) is used to correlate the tool result back to the specific call that requested it, which matters when the model makes multiple tool calls in parallel.
4. The arguments are a JSON string. Even though the model "knows" the value is a number, it serializes it as a JSON string because that is what the schema specifies. You must parse this string in your code.

After executing the tool, feed the result back:

import json

tool_call = message.tool_calls[0]
order_data = get_order_status(order_id="48291")  # Your real function

messages = [
    {"role": "system", "content": "You are an order status assistant."},
    {"role": "user", "content": "What is the status of order #48291?"},
    message,  # The assistant message containing the tool call
    {
        "role": "tool",
        "tool_call_id": tool_call.id,
        "content": json.dumps(order_data),
    },
]

final_response = openai.chat.completions.create(
    model="gpt-4o",
    messages=messages,
    tools=[...],  # Same tool definitions
)

print(final_response.choices[0].message.content)
# "Order #48291 was placed on March 5th and is currently being
#  prepared for shipment. No tracking number has been assigned yet."

Now the response is grounded in real data. The model did not invent a tracking number because the actual order record does not have one yet. The function calling protocol transformed the agent from a hallucinator into a data accessor.

6.3 Tool Definitions: The Art of JSON Schema

The tool definition is not just metadata. It is the interface contract between your agent and your code. A poorly defined tool is worse than no tool at all, because the model will call it with wrong arguments, get confusing errors, and either spiral or hallucinate a recovery.

Anatomy of a Tool Definition

Every tool definition has three critical components:

{
    "type": "function",
    "function": {
        "name": "search_products",
        "description":
            "Search the product catalog by keyword, category, or price range. "
            "Returns up to 10 matching products with name, price, and stock status. "
            "Use this when the user asks about product availability or pricing.",
        "parameters": {
            "type": "object",
            "properties": {
                "query": {
                    "type": "string",
                    "description": "Free-text search query, e.g. 'wireless headphones'"
                },
                "category": {
                    "type": "string",
                    "enum": ["electronics", "clothing", "home", "sports", "books"],
                    "description": "Filter by product category"
                },
                "max_price": {
                    "type": "number",
                    "description": "Maximum price in USD. Omit for no price filter."
                },
                "in_stock_only": {
                    "type": "boolean",
                    "description": "If true, only return products currently in stock",
                    "default": true
                }
            },
            "required": ["query"]
        }
    }
}

Name. Use verb_noun format: get_order_status, search_products, create_ticket. Avoid generic names like query or helper. The name is the first thing the model reads when deciding which tool to call, and ambiguous names cause misrouting.

Description. This is the most important field in the entire definition. Write it as if you are explaining the tool to a new developer on your team. Include what the tool does, what it returns, and when to use it. That last part — usage guidance — is what separates tools that get called correctly from tools that get called at the wrong time. Models use the description for routing decisions, not just argument formatting.

Parameters. Use the tightest schema that accurately represents the input space. If a parameter has a fixed set of valid values, use enum. If a parameter is optional, leave it out of required. Every constraint you add to the schema is a constraint the model can use to generate correct arguments.

Common Schema Pitfalls

6.4 Tool Registry Architecture

When you have three tools, you can hardcode them in a list. When you have thirty, you need a registry. When you have three hundred — which happens fast in enterprise systems — you need auto-discovery, categorization, and dynamic selection. The tool registry manages all of this.

A registry serves four functions: it stores tool definitions, validates them on registration, indexes them for fast lookup, and exports them in the format that LLM APIs expect.

from dataclasses import dataclass, field
from typing import Callable, Any
import json
import inspect
import jsonschema

@dataclass
class Tool:
    """A registered tool with its metadata and implementation."""
    name: str
    description: str
    parameters: dict
    function: Callable
    tags: list[str] = field(default_factory=list)
    requires_confirmation: bool = False
    timeout_seconds: int = 30

    def to_openai_schema(self) -> dict:
        """Export in OpenAI function calling format."""
        return {
            "type": "function",
            "function": {
                "name": self.name,
                "description": self.description,
                "parameters": self.parameters,
            }
        }


class ToolRegistry:
    """Central registry for agent tools with validation and discovery."""

    def __init__(self):
        self._tools: dict[str, Tool] = {}
        self._tag_index: dict[str, list[str]] = {}

    def register(self, tool: Tool) -> None:
        """Register a tool, validating its schema first."""
        self._validate_schema(tool)
        self._tools[tool.name] = tool
        for tag in tool.tags:
            self._tag_index.setdefault(tag, []).append(tool.name)

    def _validate_schema(self, tool: Tool) -> None:
        """Ensure the tool schema is valid JSON Schema."""
        if not tool.name.replace("_", "").isalnum():
            raise ValueError(f"Tool name must be alphanumeric: {tool.name}")
        if not tool.description or len(tool.description) < 10:
            raise ValueError(f"Tool '{tool.name}' needs a meaningful description")
        try:
            jsonschema.Draft7Validator.check_schema(tool.parameters)
        except jsonschema.SchemaError as e:
            raise ValueError(f"Invalid schema for '{tool.name}': {e.message}")

    def get(self, name: str) -> Tool:
        """Look up a tool by name."""
        if name not in self._tools:
            raise KeyError(f"No tool registered with name '{name}'")
        return self._tools[name]

    def get_by_tag(self, tag: str) -> list[Tool]:
        """Get all tools matching a tag."""
        names = self._tag_index.get(tag, [])
        return [self._tools[n] for n in names]

    def export_schemas(self, tags: list[str] | None = None) -> list[dict]:
        """Export tool schemas for the LLM API, optionally filtered by tags."""
        if tags:
            tools = set()
            for tag in tags:
                tools.update(self._tag_index.get(tag, []))
            return [self._tools[n].to_openai_schema() for n in tools]
        return [t.to_openai_schema() for t in self._tools.values()]

    def list_tools(self) -> list[str]:
        """List all registered tool names."""
        return list(self._tools.keys())

Design decisions worth examining:

• Schema validation on registration. If a tool has an invalid JSON Schema, it fails at registration time, not at runtime when the LLM tries to call it.
• Tag-based indexing. Tags like "database", "search", or "admin" enable filtering tools by category. When an agent is handling a search query, you can pass only search-tagged tools to the LLM, reducing confusion and token usage.
• Confirmation flag. Tools marked requires_confirmation=True (like delete_account or send_payment) signal to the dispatcher that a human approval step is needed before execution.
• Timeout. Each tool has a timeout. A web scraping tool might need 30 seconds; a calculator needs 1. Without per-tool timeouts, a slow tool can hang your entire agent loop.

Figure 6.1 — Tool registry architecture. Definitions flow in from the left, the registry validates and indexes them, the dispatcher routes LLM tool calls to implementations, and the result parser serializes outputs back to the conversation.

Auto-Discovery with Decorators

Manually constructing Tool objects for every function is tedious and error-prone. A better approach: let developers register tools using a decorator that extracts metadata from the function itself.

import functools
from typing import get_type_hints

def tool(
    name: str = None,
    description: str = None,
    tags: list[str] = None,
    requires_confirmation: bool = False,
):
    """Decorator that auto-registers a function as an agent tool."""
    def decorator(func: Callable) -> Callable:
        tool_name = name or func.__name__
        tool_desc = description or func.__doc__ or ""
        hints = get_type_hints(func)

        # Build JSON Schema from type hints
        params = _build_schema_from_hints(func, hints)

        tool_obj = Tool(
            name=tool_name,
            description=tool_desc.strip(),
            parameters=params,
            function=func,
            tags=tags or [],
            requires_confirmation=requires_confirmation,
        )

        # Register with the global registry
        global_registry.register(tool_obj)

        @functools.wraps(func)
        def wrapper(*args, **kwargs):
            return func(*args, **kwargs)

        wrapper._tool = tool_obj
        return wrapper

    return decorator


def _build_schema_from_hints(func, hints) -> dict:
    """Convert Python type hints into JSON Schema."""
    sig = inspect.signature(func)
    properties = {}
    required = []

    type_map = {str: "string", int: "integer", float: "number", bool: "boolean"}

    for param_name, param in sig.parameters.items():
        if param_name == "self":
            continue
        python_type = hints.get(param_name, str)
        json_type = type_map.get(python_type, "string")

        properties[param_name] = {"type": json_type}

        # Use docstring parsing or annotations for descriptions
        if param.default is inspect.Parameter.empty:
            required.append(param_name)

    return {"type": "object", "properties": properties, "required": required}

Now registering a tool is a one-liner:

@tool(tags=["orders"], description=(
    "Look up the current status of a customer order. "
    "Returns order date, status, tracking info, and shipping address. "
    "Use when a customer asks about their order."
))
def get_order_status(order_id: str) -> dict:
    """Fetch order status from the database."""
    return db.orders.find_one({"order_id": order_id})

The decorator extracts the function name, builds a JSON Schema from the type hints, and registers it with the global registry. When the agent starts up, every decorated function is already available. Add a new file with a decorated function, restart the agent, and the tool appears automatically.

6.5 The Dispatcher

The dispatcher is the bridge between the LLM's tool call requests and the actual function implementations. It looks up the requested tool in the registry, validates the arguments against the schema, executes the function with appropriate error handling and timeouts, and returns the result in a format the LLM can consume.

import asyncio
import json
import jsonschema

class ToolDispatcher:
    """Executes tool calls from the LLM with validation and error handling."""

    def __init__(self, registry: ToolRegistry):
        self.registry = registry

    async def dispatch(self, tool_call) -> dict:
        """Execute a single tool call and return the result."""
        name = tool_call.function.name
        raw_args = tool_call.function.arguments

        # Step 1: Look up the tool
        try:
            tool = self.registry.get(name)
        except KeyError:
            return self._error_result(
                tool_call.id,
                f"Unknown tool: '{name}'. Available: {self.registry.list_tools()}"
            )

        # Step 2: Parse and validate arguments
        try:
            args = json.loads(raw_args)
            jsonschema.validate(args, tool.parameters)
        except json.JSONDecodeError as e:
            return self._error_result(tool_call.id, f"Invalid JSON: {e}")
        except jsonschema.ValidationError as e:
            return self._error_result(tool_call.id, f"Invalid arguments: {e.message}")

        # Step 3: Check confirmation requirement
        if tool.requires_confirmation:
            return self._confirmation_result(tool_call.id, name, args)

        # Step 4: Execute with timeout
        try:
            result = await asyncio.wait_for(
                asyncio.to_thread(tool.function, **args),
                timeout=tool.timeout_seconds,
            )
        except asyncio.TimeoutError:
            return self._error_result(
                tool_call.id,
                f"Tool '{name}' timed out after {tool.timeout_seconds}s"
            )
        except Exception as e:
            return self._error_result(
                tool_call.id,
                f"Tool '{name}' raised an error: {type(e).__name__}: {e}"
            )

        # Step 5: Serialize the result
        return {
            "role": "tool",
            "tool_call_id": tool_call.id,
            "content": self._serialize_result(result),
        }

    def _serialize_result(self, result, max_length: int = 4000) -> str:
        """Convert result to JSON string, truncating if needed."""
        text = json.dumps(result, default=str, indent=2)
        if len(text) > max_length:
            text = text[:max_length] + "\n... [truncated]"
        return text

    def _error_result(self, call_id: str, message: str) -> dict:
        return {
            "role": "tool",
            "tool_call_id": call_id,
            "content": json.dumps({"error": message}),
        }

    def _confirmation_result(self, call_id: str, name: str, args: dict) -> dict:
        return {
            "role": "tool",
            "tool_call_id": call_id,
            "content": json.dumps({
                "status": "confirmation_required",
                "tool": name,
                "args": args,
                "message": f"Tool '{name}' requires human confirmation before execution."
            }),
        }

The five-step dispatch pipeline — lookup, validate, confirm, execute, serialize — handles every failure mode that commonly derails agents:

• Unknown tool names get a clear error listing available tools, so the model can self-correct.
• Malformed arguments get schema-specific error messages (e.g., "missing required field"), which are far more useful to the model than a Python traceback.
• Dangerous operations pause for human confirmation rather than executing automatically.
• Hanging tools time out cleanly instead of blocking the agent loop forever.
• Runtime exceptions are caught and returned as structured errors rather than crashing the process.

6.6 Result Parsing and Context Management

Tool results vary wildly in size and structure. A calculator returns a single number. A database query might return fifty rows of JSON. A web scraper might return twenty kilobytes of HTML. The result parser normalizes these outputs into something the LLM can actually use within its context window.

Strategies for Large Results

Truncation with summary. For results exceeding a threshold (4,000 characters is a reasonable default), truncate the data but prepend a summary: "Returned 47 results. Showing first 10. Use pagination to see more."

Projection. If the tool returns full database rows but the model only needs three fields, strip the result down before returning it. This requires knowing what the model is likely to need, which you can sometimes infer from the tool call arguments or the original user query.

Pagination. Instead of returning all results, return a page with a cursor. The model can call the tool again with page=2 if it needs more. This mirrors how humans interact with APIs and works naturally with the agent loop pattern.

class ResultParser:
    """Normalize and constrain tool results for LLM consumption."""

    def __init__(self, max_chars: int = 4000, max_items: int = 20):
        self.max_chars = max_chars
        self.max_items = max_items

    def parse(self, result: Any, tool_name: str) -> str:
        """Convert raw tool output to an LLM-friendly string."""
        if result is None:
            return json.dumps({"result": None, "note": "Tool returned no data"})

        if isinstance(result, list) and len(result) > self.max_items:
            truncated = result[:self.max_items]
            return json.dumps({
                "results": truncated,
                "total_count": len(result),
                "showing": self.max_items,
                "note": f"Showing first {self.max_items} of {len(result)} results"
            }, default=str, indent=2)

        serialized = json.dumps(result, default=str, indent=2)
        if len(serialized) > self.max_chars:
            return serialized[:self.max_chars] + (
                f'\n... [truncated at {self.max_chars} chars. '
                f'Full result is {len(serialized)} chars]'
            )
        return serialized

6.7 Dynamic Tool Selection

When an agent has access to dozens of tools, passing all their schemas to the LLM on every turn wastes tokens and degrades selection accuracy. The model must read and understand every tool definition to decide which one to call. With 50 tools, that is thousands of tokens of schema boilerplate on every API call.

Dynamic tool selection gives the model only the tools relevant to the current turn.

Tag-Based Filtering

The simplest approach. Analyze the user's query and select tool categories:

def select_tools_by_intent(query: str, registry: ToolRegistry) -> list[dict]:
    """Use a lightweight classifier to select relevant tool categories."""
    intent = classify_intent(query)  # Returns tags like "search", "orders", etc.
    return registry.export_schemas(tags=intent.tags)

Embedding-Based Selection

For larger registries, embed the tool descriptions and the user query in the same vector space. Select the top-k most similar tools:

import numpy as np

class EmbeddingToolSelector:
    """Select tools using semantic similarity."""

    def __init__(self, registry: ToolRegistry, embed_fn: Callable):
        self.registry = registry
        self.embed_fn = embed_fn
        self._index = self._build_index()

    def _build_index(self) -> dict:
        index = {}
        for name in self.registry.list_tools():
            tool = self.registry.get(name)
            text = f"{tool.name}: {tool.description}"
            index[name] = self.embed_fn(text)
        return index

    def select(self, query: str, top_k: int = 5) -> list[dict]:
        query_embedding = self.embed_fn(query)
        scores = {
            name: np.dot(query_embedding, emb)
            for name, emb in self._index.items()
        }
        top_names = sorted(scores, key=scores.get, reverse=True)[:top_k]
        return [self.registry.get(n).to_openai_schema() for n in top_names]

Two-Stage Selection

For the largest registries (hundreds of tools), use a two-stage approach: first, a fast filter (tags or embeddings) narrows to ~20 candidates. Then, a small LLM call selects the final 3-5 tools from those candidates. This costs one extra API call but dramatically improves accuracy.

6.8 Tool Composition

Individual tools are atoms. Useful work usually requires molecules — sequences of tool calls that together accomplish something no single tool can.

Agent-Driven Composition

The agent decides on its own to chain tools together. You give it search_products, check_inventory, and calculate_shipping. When a user asks to ship the cheapest wireless headphones to Austin, TX by Friday, the agent reasons through the sequence: search for headphones, filter by price, check inventory for the cheapest, calculate shipping to Austin with a Friday deadline. Each step uses one tool, and the agent plans the pipeline dynamically.

This is the default pattern in ReAct agents, and it works well for novel combinations. The downside is that each tool call is a separate LLM turn, adding latency and cost.

Pre-Composed Tools

For common sequences, create a composite tool that executes the entire pipeline in one call:

@tool(tags=["orders", "shipping"], description=(
    "Find the cheapest product matching a query that can be shipped to a "
    "destination by a deadline. Returns product details, price, and "
    "shipping options. Use when the user needs availability + shipping in one step."
))
def find_and_ship(query: str, destination: str, deadline: str) -> dict:
    """Composite: search + inventory + shipping in one step."""
    products = search_products(query=query)
    if not products:
        return {"error": "No products found matching query"}

    cheapest = min(products, key=lambda p: p["price"])
    stock = check_inventory(product_id=cheapest["id"])
    if not stock["available"]:
        return {"error": f"'{cheapest['name']}' is out of stock"}

    shipping = calculate_shipping(
        product_id=cheapest["id"],
        destination=destination,
        deadline=deadline,
    )
    return {
        "product": cheapest,
        "in_stock": True,
        "shipping": shipping,
    }

Pre-composed tools trade flexibility for efficiency. The agent makes one tool call instead of three, saving two LLM round-trips. The composition logic is hardcoded: if the user's request does not quite match the pre-composed pipeline, the agent cannot adapt.

In practice, use both. Pre-compose the common workflows (80% of requests), and let the agent compose dynamically for the long tail.

Figure 6.2 — Function calling sequence. The user message flows to the LLM, which emits a structured tool call. The host application executes the real function, returns the result, and the LLM composes a grounded final response.

6.9 Error Handling Patterns

Tools fail. APIs go down, databases time out, and users provide invalid inputs. There are three strategies, and production agents typically use all three.

Strategy 1: Structured Error Return

Return the error as a structured message the model can understand and act on:

# The model called search_products(query="headphones", max_price="fifty")
# The dispatcher validated against schema and returned:
{"error": "Invalid arguments: 'max_price' must be a number, got string 'fifty'"}

# The model reads this error and self-corrects:
# Next tool call: search_products(query="headphones", max_price=50)

This works because the error message is specific. The model knows exactly which parameter was wrong and what type it should be. Vague errors like "Bad request" give the model nothing to work with.

Strategy 2: Retry with Backoff

For transient failures (network timeouts, rate limits), implement automatic retries in the dispatcher. The model does not need to know about retries — they happen transparently:

async def dispatch_with_retry(self, tool_call, max_retries: int = 2) -> dict:
    """Retry transient failures automatically."""
    for attempt in range(max_retries + 1):
        result = await self.dispatch(tool_call)
        content = json.loads(result["content"])

        if "error" not in content:
            return result
        if not self._is_transient(content["error"]):
            return result  # Non-transient error, return immediately
        if attempt < max_retries:
            await asyncio.sleep(2 ** attempt)  # Exponential backoff

    return result  # Return the last error after all retries exhausted

def _is_transient(self, error_msg: str) -> bool:
    """Classify whether an error is transient (retryable)."""
    transient_patterns = ["timeout", "rate limit", "503", "connection refused"]
    return any(p in error_msg.lower() for p in transient_patterns)

Strategy 3: Fallback Tools

When the primary tool fails, offer an alternative:

@tool(tags=["orders"], description=(
    "Fallback order lookup using the read replica. Use only when "
    "get_order_status fails. Data may be up to 5 minutes stale."
))
def get_order_status_fallback(order_id: str) -> dict:
    """Read from the replica database."""
    result = db_replica.orders.find_one({"order_id": order_id})
    if result:
        result["_note"] = "Data from read replica, may be up to 5 minutes stale"
    return result

The model learns about fallbacks from the tool descriptions. When it sees that get_order_status failed, it reads the available tools and discovers the fallback. The description tells it when to use the fallback and what limitations to expect.

6.10 Security Considerations

Tools extend the agent's capabilities into the real world. That makes them the primary attack surface. Every tool is a potential vector for prompt injection, data exfiltration, and unauthorized actions.

Input Validation

Never trust the arguments the model provides. The JSON Schema validation in the dispatcher catches type errors, but it cannot catch semantic attacks. Consider a file-reading tool:

# DANGEROUS: no path validation
@tool(description="Read a file from the project directory.")
def read_file(path: str) -> str:
    return open(path).read()

# The model might be prompted to call:
# read_file(path="/etc/passwd")
# read_file(path="../../secrets/api_keys.json")

# SAFE: validate and constrain paths
@tool(description="Read a file from the project directory.")
def read_file(path: str) -> str:
    resolved = Path(PROJECT_DIR, path).resolve()
    if not resolved.is_relative_to(PROJECT_DIR):
        raise ValueError("Path must be within the project directory")
    return resolved.read_text()

Sandboxing

Tools that execute arbitrary code — like code interpreters or shell commands — must run in sandboxed environments. At minimum:

• Process isolation. Run in a subprocess or container with no network access and limited filesystem access.
• Resource limits. Cap CPU time, memory usage, and disk writes. A malicious prompt could instruct the agent to run an infinite loop or fill the disk.
• No credential access. The sandbox must not have access to environment variables, credential files, or API keys that the host application uses.

Least Privilege

Give each tool only the permissions it needs. A tool that reads orders should not have write access to the orders database. A tool that searches products should not have access to customer data. This maps directly to the tag system in the registry.

Audit Logging

Log every tool call, its arguments, and its result. When something goes wrong, you need a complete trace of every action the agent took:

import logging
import time

logger = logging.getLogger("tool_audit")

class AuditingDispatcher(ToolDispatcher):
    """Dispatcher that logs every tool call for security audit."""

    async def dispatch(self, tool_call) -> dict:
        start = time.monotonic()
        logger.info(
            "tool_call_start",
            extra={
                "tool": tool_call.function.name,
                "args": tool_call.function.arguments,
                "call_id": tool_call.id,
            }
        )

        result = await super().dispatch(tool_call)

        logger.info(
            "tool_call_end",
            extra={
                "call_id": tool_call.id,
                "duration_ms": (time.monotonic() - start) * 1000,
                "result_length": len(result["content"]),
                "has_error": "error" in result["content"],
            }
        )
        return result

6.11 Putting It All Together

Here is how the complete tool pipeline works in an agent loop — registry, dispatcher, dynamic selection, and error handling combined into a working system:

import openai

class ToolAgent:
    """Agent with full tool pipeline: registry, selection, dispatch."""

    def __init__(self, registry: ToolRegistry, model: str = "gpt-4o"):
        self.registry = registry
        self.dispatcher = AuditingDispatcher(registry)
        self.selector = EmbeddingToolSelector(registry, embed_fn=get_embedding)
        self.model = model

    async def run(self, user_message: str, max_turns: int = 10) -> str:
        messages = [
            {"role": "system", "content": SYSTEM_PROMPT},
            {"role": "user", "content": user_message},
        ]

        for turn in range(max_turns):
            # Dynamic tool selection based on conversation context
            relevant_tools = self.selector.select(
                query=messages[-1]["content"], top_k=8
            )

            response = openai.chat.completions.create(
                model=self.model,
                messages=messages,
                tools=relevant_tools if relevant_tools else None,
            )

            message = response.choices[0].message
            messages.append(message)

            # If no tool calls, the agent is done
            if not message.tool_calls:
                return message.content

            # Execute all tool calls (possibly in parallel)
            tool_results = []
            for tc in message.tool_calls:
                result = await self.dispatcher.dispatch_with_retry(tc)
                tool_results.append(result)

            messages.extend(tool_results)

        return "I was unable to complete the task within the turn limit."

This is a production-ready skeleton. The agent selects relevant tools per turn, executes them with retries and auditing, and loops until the model is satisfied or the turn limit is reached. Every component — registry, dispatcher, selector, result parser — is independently testable and swappable.

Project: Tool Registry Framework

Build an extensible tool registry system with auto-discovery. Your framework should support: decorator-based registration, JSON Schema validation, tag-based and embedding-based tool selection, a five-step dispatcher with retry logic, audit logging, and at least five working tools demonstrating different patterns (read, write, search, compute, external API).

Requirements

1. Registry. Implement ToolRegistry with registration, validation, tag indexing, and schema export.
2. Decorator. Create a @tool decorator that auto-registers functions with schemas inferred from type hints.
3. Dispatcher. Build ToolDispatcher with the five-step pipeline: lookup, validate, confirm, execute, serialize.
4. Dynamic selection. Implement at least one selection strategy (tag-based or embedding-based).
5. Error handling. Structured errors, retry with backoff for transient failures, at least one fallback tool.
6. Security. Input validation, path sandboxing for file tools, audit logging.
7. Agent integration. Wire the registry into a ReAct agent loop that uses tools to answer real queries.

Domain Variants

Exercises