Custom Tools Example

Add your own tools alongside built-in toolsets.

Source Code

examples/custom_tools.py

Overview

This example demonstrates:

• Defining custom tool functions
• Accessing dependencies in custom tools
• Combining custom logic with built-in file operations
• Tool return values and documentation

Full Example

Python

"""Example adding custom tools to the agent."""

import asyncio
from datetime import datetime

from pydantic_ai import RunContext

from pydantic_deep import DeepAgentDeps, StateBackend, create_deep_agent


# Define custom tools as functions
async def get_current_time(ctx: RunContext[DeepAgentDeps]) -> str:
    """Get the current date and time.

    Returns:
        Current timestamp in ISO format.
    """
    return datetime.now().isoformat()


async def log_message(
    ctx: RunContext[DeepAgentDeps],
    message: str,
    level: str = "INFO",
) -> str:
    """Log a message to /logs/agent.log.

    Args:
        message: The message to log.
        level: Log level (INFO, WARNING, ERROR).
    """
    timestamp = datetime.now().isoformat()
    log_entry = f"[{timestamp}] [{level}] {message}\n"

    # Use the backend to append to log file
    backend = ctx.deps.backend

    # Read existing log
    existing = backend.read("/logs/agent.log")
    if "Error:" in existing:
        # File doesn't exist, create it
        content = log_entry
    else:
        # Extract content (remove line numbers)
        lines = []
        for line in existing.split("\n"):
            if "\t" in line:
                lines.append(line.split("\t", 1)[1])
        content = "\n".join(lines) + log_entry

    backend.write("/logs/agent.log", content)

    return f"Logged: {log_entry.strip()}"


async def analyze_code_complexity(
    ctx: RunContext[DeepAgentDeps],
    file_path: str,
) -> str:
    """Analyze the complexity of a Python file.

    Args:
        file_path: Path to the Python file to analyze.

    Returns:
        Complexity analysis report.
    """
    content = ctx.deps.backend.read(file_path)

    if "Error:" in content:
        return content

    # Simple complexity metrics
    lines = content.split("\n")
    total_lines = len(lines)

    # Count various elements (simple heuristics)
    functions = sum(1 for line in lines if "def " in line)
    classes = sum(1 for line in lines if "class " in line)
    imports = sum(1 for line in lines if line.strip().startswith(("import ", "from ")))
    comments = sum(1 for line in lines if "#" in line)

    return f"""Code Complexity Analysis for {file_path}:
- Total lines: {total_lines}
- Functions: {functions}
- Classes: {classes}
- Imports: {imports}
- Comment lines: {comments}
- Code density: {(total_lines - comments) / max(total_lines, 1):.1%}
"""


async def main():
    # Create agent with custom tools
    agent = create_deep_agent(
        model="openai:gpt-4.1",
        instructions="""
        You are a development assistant with custom tools:
        - get_current_time: Get the current timestamp
        - log_message: Log messages to /logs/agent.log
        - analyze_code_complexity: Analyze Python file complexity

        Use these tools along with the built-in filesystem tools.
        Always log important actions.
        """,
        tools=[
            get_current_time,
            log_message,
            analyze_code_complexity,
        ],
    )

    deps = DeepAgentDeps(backend=StateBackend())

    # Run the agent
    result = await agent.run(
        """
        1. Log that we're starting a new task
        2. Create a Python module at /src/calculator.py with add, subtract, multiply functions
        3. Analyze the complexity of the created file
        4. Log the completion with the complexity summary
        5. Get the current time and save a summary to /summary.txt
        """,
        deps=deps,
    )

    print("Agent output:")
    print(result.output)

    # Show the log file
    print("\n" + "=" * 50)
    print("Log file contents:")
    log_content = deps.backend.read("/logs/agent.log")
    print(log_content)


if __name__ == "__main__":
    asyncio.run(main())

Running the Example

Bash

export OPENAI_API_KEY=your-api-key
uv run python examples/custom_tools.py

Expected Output

Text Only

Agent output:
I've completed all the tasks:

1. Logged the start of the task
2. Created /src/calculator.py with add, subtract, and multiply functions
3. Analyzed the code complexity:
   - 15 total lines
   - 3 functions
   - Good code density
4. Logged the completion
5. Saved summary to /summary.txt with current timestamp

==================================================
Log file contents:
     1  [2024-01-15T10:30:00] [INFO] Starting new task: create calculator module
     2  [2024-01-15T10:30:05] [INFO] Task completed. Code complexity: 3 functions, 15 lines

Key Concepts

Tool Function Signature

Python

async def my_tool(
    ctx: RunContext[DeepAgentDeps],  # Required: access to dependencies
    param1: str,                      # Required parameter
    param2: int = 10,                 # Optional parameter with default
) -> str:                             # Return type
    """Tool description shown to the agent.

    Args:
        param1: Description of param1.
        param2: Description of param2.

    Returns:
        Description of what the tool returns.
    """
    # Access dependencies
    backend = ctx.deps.backend
    todos = ctx.deps.todos

    # Your logic here
    return "result"

Accessing Dependencies

Python

async def my_tool(ctx: RunContext[DeepAgentDeps]) -> str:
    # Access the backend for file operations
    content = ctx.deps.backend.read("/some/file.txt")
    ctx.deps.backend.write("/output.txt", "result")

    # Access todos
    for todo in ctx.deps.todos:
        print(todo.content)

    # Access uploaded files metadata
    for path, info in ctx.deps.uploads.items():
        print(f"{path}: {info['size']} bytes")

    return "done"

Registering Tools

Python

agent = create_deep_agent(
    tools=[
        my_tool,           # Function reference
        another_tool,
        yet_another_tool,
    ],
)

Variations

Sync Tools

Tools can be synchronous if they don't need async operations:

Python

def get_version(ctx: RunContext[DeepAgentDeps]) -> str:
    """Get the application version."""
    return "1.0.0"

Tools with Complex Return Types

Python

from pydantic import BaseModel

class AnalysisResult(BaseModel):
    file_path: str
    lines: int
    complexity_score: float
    issues: list[str]

async def analyze_file(
    ctx: RunContext[DeepAgentDeps],
    path: str,
) -> AnalysisResult:
    """Analyze a file and return structured results."""
    content = ctx.deps.backend.read(path)
    # ... analysis logic ...
    return AnalysisResult(
        file_path=path,
        lines=100,
        complexity_score=0.7,
        issues=["TODO found on line 42"],
    )

Tools with External APIs

Python

import httpx

async def fetch_weather(
    ctx: RunContext[DeepAgentDeps],
    city: str,
) -> str:
    """Get current weather for a city."""
    async with httpx.AsyncClient() as client:
        response = await client.get(
            f"https://api.weather.com/v1/current?city={city}"
        )
        data = response.json()
        return f"Weather in {city}: {data['temp']}°C, {data['condition']}"

Combining with Toolsets

Python

from pydantic_ai.toolsets import FunctionToolset

# Create a toolset from multiple functions
my_toolset = FunctionToolset([
    get_current_time,
    log_message,
    analyze_code_complexity,
])

agent = create_deep_agent(
    toolsets=[my_toolset],  # Add as a toolset
)

Best Practices

1. Clear docstrings - The agent uses docstrings to understand what tools do
2. Type hints - Help the agent understand parameter types
3. Descriptive names - analyze_code_complexity vs analyze
4. Error handling - Return helpful error messages, don't raise exceptions
5. Stateless - Tools should not maintain state between calls

Next Steps

• Subagents - Delegate to specialized agents
• Skills - Package capabilities as skills
• API: Toolsets - Toolset reference