PydanticAI SDK Track Part 5: Function Tools & Toolsets

1. Function Tools with @agent.tool

Function tools are the primary mechanism for giving PydanticAI agents capabilities beyond text generation. The @agent.tool decorator transforms ordinary Python functions into tools that the model can invoke during a conversation. Type annotations drive the schema — parameter names, types, and docstrings are automatically extracted to form the tool description sent to the model.

1.1 Type Annotations Drive the Schema

PydanticAI inspects your function’s type hints to generate the JSON schema that the model sees. The function’s docstring becomes the tool’s description, and each parameter’s annotation defines its expected type:

from pydantic_ai import Agent

agent = Agent("openai:gpt-4o", system_prompt="You are a helpful weather assistant.")

@agent.tool
async def get_weather(city: str, units: str = "celsius") -> str:
    """Get the current weather for a city.

    Args:
        city: The city name to look up weather for.
        units: Temperature units - 'celsius' or 'fahrenheit'.
    """
    # In production, call a real weather API
    return f"The weather in {city} is 22 degrees {units} and sunny."

result = agent.run_sync("What's the weather like in London?")
print(result.output)
# Output: The weather in London is 22 degrees celsius and sunny.

1.2 RunContext for Accessing Dependencies

When your tool needs access to shared resources (database connections, API clients, configuration), use RunContext[DepsType] as the first parameter. PydanticAI automatically injects the dependencies you passed to agent.run():

from dataclasses import dataclass
from pydantic_ai import Agent, RunContext

@dataclass
class WeatherDeps:
    api_key: str
    base_url: str = "https://api.weather.example.com"

agent = Agent("openai:gpt-4o", deps_type=WeatherDeps)

@agent.tool
async def get_temperature(ctx: RunContext[WeatherDeps], city: str) -> str:
    """Get current temperature for a city using the weather API.

    Args:
        city: The city name to look up.
    """
    # Access dependencies via ctx.deps
    api_key = ctx.deps.api_key
    base_url = ctx.deps.base_url
    # In production: response = httpx.get(f"{base_url}/temp?city={city}&key={api_key}")
    return f"Temperature in {city}: 18°C (fetched from {base_url})"

deps = WeatherDeps(api_key="sk-weather-123")
result = agent.run_sync("What's the temperature in Paris?", deps=deps)
print(result.output)

2. Advanced Tool Features

2.1 Retries & Error Recovery

The retries parameter on @agent.tool tells PydanticAI how many times to retry a tool call if it raises a ModelRetry exception. This is powerful for self-correcting behavior — the error message is sent back to the model so it can fix its inputs:

from pydantic_ai import Agent, RunContext, ModelRetry

agent = Agent("openai:gpt-4o")

@agent.tool(retries=3)
async def search_database(query: str, max_results: int = 10) -> str:
    """Search the product database.

    Args:
        query: Search query string (minimum 3 characters).
        max_results: Maximum number of results to return (1-100).
    """
    if len(query) < 3:
        raise ModelRetry("Query must be at least 3 characters. Please provide a longer search term.")
    if max_results < 1 or max_results > 100:
        raise ModelRetry("max_results must be between 1 and 100. Please adjust.")

    # Simulated search
    return f"Found 5 products matching '{query}' (showing top {max_results})"

result = agent.run_sync("Find me laptops")
print(result.output)

2.2 Prepare Callbacks for Dynamic Tool Modification

The prepare callback runs before a tool is presented to the model. Use it to dynamically modify the tool’s description, hide the tool conditionally, or inject context-specific information:

from pydantic_ai import Agent, RunContext
from pydantic_ai.tools import ToolDefinition

agent = Agent("openai:gpt-4o")

async def check_admin_access(
    ctx: RunContext[dict], tool_def: ToolDefinition
) -> ToolDefinition | None:
    """Only show the delete tool if user has admin role."""
    if ctx.deps.get("role") != "admin":
        return None  # Hide tool from non-admins
    return tool_def  # Show tool as-is for admins

@agent.tool(prepare=check_admin_access)
async def delete_record(ctx: RunContext[dict], record_id: str) -> str:
    """Permanently delete a record from the database.

    Args:
        record_id: The unique identifier of the record to delete.
    """
    return f"Record {record_id} deleted successfully."

# Admin user sees the delete tool
result = agent.run_sync("Delete record ABC-123", deps={"role": "admin"})
print(result.output)

# Regular user does NOT see the delete tool
result = agent.run_sync("Delete record ABC-123", deps={"role": "viewer"})
print(result.output)  # Model will say it can't delete

2.3 Tool-Level System Prompts

Beyond the function docstring, you can provide additional context to the model about when and how to use a specific tool:

from pydantic_ai import Agent

agent = Agent(
    "openai:gpt-4o",
    system_prompt="You are a data analysis assistant."
)

@agent.tool
async def run_sql_query(query: str) -> str:
    """Execute a read-only SQL query against the analytics database.

    IMPORTANT GUIDELINES:
    - Only SELECT statements are allowed
    - Always include a LIMIT clause (max 1000 rows)
    - Available tables: users, orders, products, sessions
    - Date columns use ISO 8601 format

    Args:
        query: The SQL SELECT query to execute.
    """
    if not query.strip().upper().startswith("SELECT"):
        return "Error: Only SELECT queries are permitted."
    # Simulated execution
    return f"Query executed successfully: {query[:50]}... (3 rows returned)"

result = agent.run_sync("How many orders were placed last week?")
print(result.output)

3. Toolsets: Grouped Tool Collections

When you have multiple related tools (e.g., all database operations, all file system operations), group them into a Toolset. Toolsets are reusable bundles you can attach to any agent without redefining each tool:

from pydantic_ai import Agent
from pydantic_ai.tools import Toolset

# Create a reusable toolset for database operations
db_toolset = Toolset()

@db_toolset.tool
async def list_tables() -> str:
    """List all available database tables."""
    return "Tables: users, orders, products, sessions, analytics"

@db_toolset.tool
async def describe_table(table_name: str) -> str:
    """Get the schema of a database table.

    Args:
        table_name: Name of the table to describe.
    """
    schemas = {
        "users": "id (int), name (str), email (str), created_at (datetime)",
        "orders": "id (int), user_id (int), total (float), status (str)",
    }
    return schemas.get(table_name, f"Table '{table_name}' not found.")

@db_toolset.tool
async def query_table(table_name: str, limit: int = 10) -> str:
    """Query rows from a table.

    Args:
        table_name: The table to query.
        limit: Maximum rows to return.
    """
    return f"Returning {limit} rows from {table_name}"

# Attach the toolset to an agent
agent = Agent("openai:gpt-4o", toolsets=[db_toolset])

result = agent.run_sync("What tables are available and what does the users table look like?")
print(result.output)

3.1 Composing Multiple Toolsets

Agents can combine multiple toolsets for rich capability sets. Each toolset remains independently testable and reusable:

from pydantic_ai import Agent
from pydantic_ai.tools import Toolset

# File system toolset
fs_toolset = Toolset()

@fs_toolset.tool
async def read_file(path: str) -> str:
    """Read contents of a file.

    Args:
        path: The file path to read.
    """
    return f"Contents of {path}: [file data here]"

@fs_toolset.tool
async def list_directory(path: str = ".") -> str:
    """List files in a directory.

    Args:
        path: Directory path to list.
    """
    return f"Files in {path}: main.py, utils.py, config.yaml"

# HTTP toolset
http_toolset = Toolset()

@http_toolset.tool
async def http_get(url: str) -> str:
    """Make an HTTP GET request.

    Args:
        url: The URL to fetch.
    """
    return f"Response from {url}: 200 OK, body=[...]"

# Combine both toolsets into one agent
agent = Agent(
    "openai:gpt-4o",
    system_prompt="You are a development assistant with file and HTTP access.",
    toolsets=[fs_toolset, http_toolset],
)

result = agent.run_sync("Read the config.yaml file and then fetch the API status endpoint")
print(result.output)

4. Deferred Tools

Deferred tools have their schema sent to the model (so it knows they exist), but execution is deferred — the tool isn’t actually called during the agent run. Instead, the tool call information is returned to your application code for later execution. This is ideal for expensive operations that need user confirmation:

from pydantic_ai import Agent

agent = Agent("openai:gpt-4o", system_prompt="You help users manage their account.")

@agent.tool(defer=True)
async def delete_account(user_id: str, reason: str) -> str:
    """Permanently delete a user account. This action cannot be undone.

    Args:
        user_id: The user ID to delete.
        reason: Reason for account deletion.
    """
    # This won't execute during agent.run() — it's deferred
    return f"Account {user_id} deleted. Reason: {reason}"

result = agent.run_sync("I want to delete my account because I'm switching services. My ID is USR-456.")

# Check for deferred tool calls
for call in result.deferred_tool_calls:
    print(f"Deferred tool: {call.tool_name}")
    print(f"Arguments: {call.args}")
    # In production: show confirmation UI, then execute if confirmed

4.1 Confirmation Pattern with Deferred Tools

A common pattern is to present deferred tool calls to the user for confirmation before executing:

from pydantic_ai import Agent

agent = Agent("openai:gpt-4o", system_prompt="You help manage cloud infrastructure.")

@agent.tool(defer=True)
async def terminate_instance(instance_id: str, force: bool = False) -> str:
    """Terminate a cloud compute instance.

    Args:
        instance_id: The instance ID to terminate.
        force: If True, force-stop without graceful shutdown.
    """
    action = "force-terminated" if force else "terminated"
    return f"Instance {instance_id} {action}."

result = agent.run_sync("Shut down instance i-abc123")

# Application-level confirmation flow
for call in result.deferred_tool_calls:
    print(f"\n⚠️  The agent wants to: {call.tool_name}")
    print(f"   Arguments: {call.args}")
    confirm = input("   Proceed? (yes/no): ")
    if confirm.lower() == "yes":
        # Execute the deferred tool
        output = await call.execute()
        print(f"   Result: {output}")
    else:
        print("   Action cancelled.")

5. Tool Design Patterns

Well-designed tools follow the single-responsibility principle: each tool does one thing clearly. This helps the model select the right tool and reduces confusion in multi-tool scenarios.

5.1 Error Handling: ToolError vs ModelRetry

PydanticAI provides two exception types for tool error handling. ModelRetry tells the model to try again with different inputs. ToolError reports a permanent failure back to the model so it can inform the user:

from pydantic_ai import Agent, RunContext, ModelRetry
from pydantic_ai.exceptions import ToolError

agent = Agent("openai:gpt-4o")

@agent.tool(retries=2)
async def fetch_user_profile(user_id: str) -> str:
    """Fetch a user's profile by their ID.

    Args:
        user_id: The user ID (format: USR-XXXX).
    """
    # Validate format — model can retry with correct format
    if not user_id.startswith("USR-"):
        raise ModelRetry(
            f"Invalid user ID format '{user_id}'. "
            "User IDs must start with 'USR-' followed by digits (e.g., USR-1234)."
        )

    # Simulate a permanent failure (user not found)
    known_users = {"USR-1234", "USR-5678"}
    if user_id not in known_users:
        raise ToolError(f"User {user_id} not found in the database.")

    return f"Profile for {user_id}: name=Alice, role=admin, active=true"

result = agent.run_sync("Get the profile for user USR-1234")
print(result.output)

A robust tool should format its results for optimal model understanding — structured, concise, and actionable:

from pydantic_ai import Agent
import json

agent = Agent("openai:gpt-4o", system_prompt="You are a sales analytics assistant.")

@agent.tool
async def get_sales_summary(period: str, region: str = "all") -> str:
    """Get sales summary for a given time period.

    Args:
        period: Time period - 'today', 'week', 'month', or 'quarter'.
        region: Filter by region or 'all' for global.
    """
    # Return structured data the model can reason about
    summary = {
        "period": period,
        "region": region,
        "total_revenue": 125000.50,
        "total_orders": 342,
        "avg_order_value": 365.50,
        "top_product": "Enterprise License",
        "growth_vs_previous": "+12.3%",
    }
    return json.dumps(summary, indent=2)

result = agent.run_sync("How are sales doing this month in EMEA?")
print(result.output)