The Gap Between Schema and Reality
A well-written tool schema is your primary defence against model hallucination in tool arguments. Vague schemas produce vague arguments. Precise schemas produce precise arguments.
Here’s the transformation from a bad schema to a good one for the same tool:
Bad schema:
{
"name": "search_orders",
"description": "Search orders",
"parameters": {
"properties": {
"customer": {"type": "string"},
"date": {"type": "string"},
"status": {"type": "string"}
}
}
}Good schema:
{
"name": "search_orders",
// Clear "when to use" description helps the model pick the right tool
"description": "Search customer orders in the database. Returns a list of matching orders. Use this when the user asks about their orders, delivery status, or order history.",
"parameters": {
"type": "object",
"properties": {
"customer_email": {
"type": "string",
"format": "email",
// Concrete example prevents the model from guessing the format
"description": "Customer's email address (lowercase). Example: [email protected]"
},
"date_from": {
"type": "string",
// Regex pattern constrains the model to exactly YYYY-MM-DD, no "January 1st" variants
"pattern": "^\\d{4}-\\d{2}-\\d{2}$",
"description": "Start date in ISO 8601 format (YYYY-MM-DD). Example: 2024-01-01"
},
"date_to": {
"type": "string",
"pattern": "^\\d{4}-\\d{2}-\\d{2}$",
// Cross-field constraint stated in prose, JSON Schema can't enforce this, but the model reads it
"description": "End date in ISO 8601 format (YYYY-MM-DD). Must be >= date_from."
},
"status": {
"type": "string",
// enum eliminates free-text hallucination, the model must pick one of these exact strings
"enum": ["pending", "shipped", "delivered", "cancelled"],
"description": "Filter by order status. Omit to return all statuses."
},
"limit": {
"type": "integer",
// minimum/maximum enforce numeric bounds at the schema level
"minimum": 1,
"maximum": 50,
"default": 10,
"description": "Maximum number of orders to return (1-50)."
}
},
// required list tells the model which fields it must always provide
"required": ["customer_email"],
// additionalProperties: false rejects any key not defined above, prevents hallucinated extra fields
"additionalProperties": false
}
}The good schema gives the model everything it needs: exact formats with examples, enum constraints for status (no guessing), and a clear description of when to use this tool.
Pydantic-based Tool Argument Validation
Even a well-defined schema can’t catch every bad argument the model might generate, semantic validation (e.g. date range order) must happen in code. Pydantic lets you express these rules as a model class that runs before any database call.
# src/agents/tools/order_search.py
from datetime import date, datetime
from pydantic import BaseModel, EmailStr, Field, field_validator, model_validator
from typing import Literal
class OrderSearchArgs(BaseModel):
# EmailStr validates format at the type level: no regex needed in the function body
customer_email: EmailStr
# date type means Pydantic auto-parses "2024-01-15" strings into Python date objects
date_from: date | None = None
date_to: date | None = None
# Literal enforces the same enum constraint as the JSON schema, now in Python
status: Literal["pending", "shipped", "delivered", "cancelled"] | None = None
# Field(ge=1, le=50) mirrors the JSON schema minimum/maximum: double enforcement
limit: int = Field(default=10, ge=1, le=50)
@model_validator(mode='after')
def validate_date_range(self) -> 'OrderSearchArgs':
# Cross-field validation: JSON Schema cannot enforce this, Pydantic can
if self.date_from and self.date_to:
if self.date_to < self.date_from:
raise ValueError(
f"date_to ({self.date_to}) must be >= date_from ({self.date_from})"
)
return self
async def search_orders(
customer_email: str,
date_from: str | None = None,
date_to: str | None = None,
status: str | None = None,
limit: int = 10,
) -> dict:
"""Search customer orders. Returns validated list or error."""
# Validate arguments before hitting the database
try:
args = OrderSearchArgs(
customer_email=customer_email,
date_from=date_from,
date_to=date_to,
status=status,
limit=limit,
)
except ValueError as e:
# Return structured error: model can self-correct by reading the message and hint
return {
"success": False,
"error": "invalid_arguments",
"message": str(e),
"hint": "Check date format (YYYY-MM-DD) and ensure date_to >= date_from"
}
# Safe to proceed with validated args
try:
results = await db_search_orders(args)
return {"success": True, "orders": results, "count": len(results)}
except Exception as e:
# retriable: True signals to the model that trying again is worthwhile
return {
"success": False,
"error": "database_error",
"message": "Failed to query orders database",
"retriable": True
}Retry Logic with Tenacity
External tools fail. APIs rate-limit. Networks timeout. Use tenacity for structured retry behaviour:
# src/agents/tools/web_search.py
import asyncio
import httpx
from tenacity import (
retry,
stop_after_attempt,
wait_exponential,
retry_if_exception_type,
before_sleep_log,
)
import logging
logger = logging.getLogger(__name__)
# @retry decorator wraps the function: failures trigger automatic retries without changing call sites
@retry(
# stop_after_attempt(3): give up after 3 total tries (1 original + 2 retries)
stop=stop_after_attempt(3),
# wait_exponential: first retry waits ~1s, second ~2s, capped at 10s: avoids hammering the API
wait=wait_exponential(multiplier=1, min=1, max=10),
# only retry on transient network errors: don't retry 400 Bad Request or auth failures
retry=retry_if_exception_type((httpx.TimeoutException, httpx.HTTPStatusError)),
# logs a WARNING before each sleep so you can see retry attempts in production logs
before_sleep=before_sleep_log(logger, logging.WARNING),
)
async def search_tavily(query: str, num_results: int = 5) -> list[dict]:
"""Search web using Tavily API with automatic retries."""
async with httpx.AsyncClient(timeout=10.0) as client:
response = await client.post(
"https://api.tavily.com/search",
json={
"api_key": settings.tavily_api_key,
"query": query,
"num_results": num_results,
"search_depth": "advanced",
}
)
if response.status_code == 429: # rate limited
# Respect the server's own retry guidance before raising to trigger the tenacity retry
retry_after = int(response.headers.get("Retry-After", 5))
await asyncio.sleep(retry_after)
response.raise_for_status() # trigger retry
response.raise_for_status()
return response.json()["results"]The Error Response Pattern
When a tool fails, what you return matters. Returning a helpful error object lets the model reason about what happened and potentially adapt:
# Pattern: always return a dict with a 'success' key
# Consistent shape means the model can reliably check success before reading data
class ToolResult:
@staticmethod
def ok(data: dict | list) -> dict:
return {"success": True, "data": data}
@staticmethod
def error(
code: str,
message: str,
retriable: bool = False,
suggestion: str = "",
) -> dict:
# structured error codes let the model distinguish "bad input" from "try again later"
return {
"success": False,
"error": {"code": code, "message": message},
# retriable tells the model whether it's worth calling this tool again
"retriable": retriable,
# suggestion gives the model a concrete next action to try
"suggestion": suggestion,
}
# Usage in a tool
async def get_stock_price(ticker: str) -> dict:
# Validate input before any network call: fail fast with a clear message
if not ticker.isalpha() or len(ticker) > 5:
return ToolResult.error(
code="invalid_ticker",
message=f"'{ticker}' is not a valid stock ticker",
suggestion="Use a standard stock symbol like 'AAPL', 'MSFT', or 'GOOGL'"
)
try:
price = await fetch_price_from_api(ticker.upper())
return ToolResult.ok({"ticker": ticker.upper(), "price": price, "currency": "USD"})
except TickerNotFoundError:
# retriable=False: retrying with the same ticker won't help
return ToolResult.error(
code="ticker_not_found",
message=f"No price data found for '{ticker}'",
suggestion="Verify the ticker symbol is correct"
)
except RateLimitError:
# retriable=True: the model can try again later or switch data sources
return ToolResult.error(
code="rate_limited",
message="Price API is rate limited",
retriable=True,
suggestion="Try again in 60 seconds or use a different data source"
)Tool Timeout Guards
Always set timeouts. An agent waiting forever for a hung API call will exhaust its iteration budget:
import asyncio
async def safe_tool_call(tool_func, *args, timeout_seconds: float = 15.0, **kwargs) -> dict:
"""Execute a tool with a hard timeout."""
try:
# asyncio.wait_for raises TimeoutError if tool_func doesn't complete in time
return await asyncio.wait_for(
tool_func(*args, **kwargs),
timeout=timeout_seconds
)
except asyncio.TimeoutError:
# Return a retriable error: the model can decide whether to retry or move on
return {
"success": False,
"error": {"code": "timeout", "message": f"Tool timed out after {timeout_seconds}s"},
"retriable": True,
"suggestion": "The operation took too long. Try a more specific query."
}Checklist: Production-ready Tool Definition
Before shipping any tool to production, verify:
- Description explains when to use this tool (not just what it does)
- All parameter descriptions include format examples (
YYYY-MM-DD,ISO 4217 currency code) - Enum values are used wherever the input is constrained
-
additionalProperties: falseprevents unexpected argument keys - Input is validated with Pydantic before hitting external systems
- All code paths return a dict (never raises uncaught exceptions)
- External HTTP calls have timeouts ≤ 15s
- Rate-limited APIs have retry logic with exponential backoff
- Error returns include
retriableandsuggestionfields - The tool is unit-tested with valid inputs, invalid inputs, and simulated failures