PydanticAI SDK Track Part 6: Native, Common & Third-Party Tools

1. Native Tools (Provider-Managed)

Native tools are executed by the model provider’s infrastructure, not your local code. They run in the provider’s sandbox environment — meaning you don’t need to implement the logic yourself. The model calls them automatically when needed, and results flow back into the conversation transparently.

1.1 Code Interpreter

Code interpreter lets the model write and execute Python code within the provider’s sandboxed environment. This is powerful for data analysis, mathematical computations, and generating visualizations without requiring local execution:

from pydantic_ai import Agent
from pydantic_ai.common_tools.code_interpreter import code_interpreter_tool

agent = Agent(
    "openai:gpt-4o",
    system_prompt="You are a data analyst. Use code execution for calculations.",
    tools=[code_interpreter_tool()],
)

result = agent.run_sync(
    "Calculate the compound interest on $10,000 at 5% annual rate "
    "over 10 years with monthly compounding. Show the formula and result."
)
print(result.output)

1.2 Web Search

Web search tools give agents access to real-time information beyond their training data. The provider handles the search infrastructure — you just enable it:

from pydantic_ai import Agent
from pydantic_ai.common_tools.web_search import web_search_tool

agent = Agent(
    "openai:gpt-4o",
    system_prompt="You answer questions using the latest available information.",
    tools=[web_search_tool()],
)

result = agent.run_sync("What are the latest developments in quantum computing this month?")
print(result.output)

2. Common Tools Library

PydanticAI ships with a set of pre-built common tools for frequent operations. These are local tools (they run in your process) but save you from writing boilerplate implementations:

2.1 File System Tools

from pydantic_ai import Agent
from pydantic_ai.common_tools.filesystem import (
    read_file_tool,
    write_file_tool,
    list_directory_tool,
)

agent = Agent(
    "openai:gpt-4o",
    system_prompt=(
        "You are a file management assistant. "
        "You can read, write, and list files in the workspace."
    ),
    tools=[
        read_file_tool(base_path="./workspace"),
        write_file_tool(base_path="./workspace"),
        list_directory_tool(base_path="./workspace"),
    ],
)

result = agent.run_sync("List all Python files in the workspace and show me the contents of main.py")
print(result.output)

2.2 HTTP Request Tools

from pydantic_ai import Agent
from pydantic_ai.common_tools.http import http_get_tool, http_post_tool

agent = Agent(
    "openai:gpt-4o",
    system_prompt="You can fetch data from APIs.",
    tools=[
        http_get_tool(allowed_domains=["api.github.com", "httpbin.org"]),
        http_post_tool(allowed_domains=["httpbin.org"]),
    ],
)

result = agent.run_sync("Fetch the public repos for user 'pydantic' from GitHub API")
print(result.output)

3. Third-Party Tool Integrations

The PydanticAI ecosystem supports third-party tool packages that extend agent capabilities without requiring you to build everything from scratch:

3.1 Adapting External Libraries as Tools

Any Python library can be wrapped as a PydanticAI tool. The pattern is simple: create a function with proper type annotations and docstring, then decorate it:

from pydantic_ai import Agent
import subprocess

agent = Agent("openai:gpt-4o", system_prompt="You help with Git repository management.")

@agent.tool
async def git_status(repo_path: str = ".") -> str:
    """Check the git status of a repository.

    Args:
        repo_path: Path to the git repository.
    """
    try:
        result = subprocess.run(
            ["git", "status", "--short"],
            cwd=repo_path,
            capture_output=True,
            text=True,
            timeout=10,
        )
        if result.returncode != 0:
            return f"Error: {result.stderr.strip()}"
        return result.stdout.strip() or "Working tree clean"
    except subprocess.TimeoutExpired:
        return "Error: git command timed out"
    except FileNotFoundError:
        return "Error: git is not installed or repo path not found"

@agent.tool
async def git_log(repo_path: str = ".", count: int = 5) -> str:
    """Get recent git commit log.

    Args:
        repo_path: Path to the git repository.
        count: Number of recent commits to show (max 20).
    """
    count = min(count, 20)  # Safety limit
    try:
        result = subprocess.run(
            ["git", "log", f"--oneline", f"-{count}"],
            cwd=repo_path,
            capture_output=True,
            text=True,
            timeout=10,
        )
        return result.stdout.strip() or "No commits found"
    except Exception as e:
        return f"Error: {str(e)}"

result = agent.run_sync("What's the current git status and show me the last 3 commits?")
print(result.output)

4. Combining Tool Types

The real power of PydanticAI emerges when you mix function tools, native tools, and third-party tools in a single agent. The model intelligently selects which tool to use based on the task:

from pydantic_ai import Agent
from pydantic_ai.common_tools.code_interpreter import code_interpreter_tool
from pydantic_ai.common_tools.web_search import web_search_tool

agent = Agent(
    "openai:gpt-4o",
    system_prompt=(
        "You are a research assistant with access to web search, "
        "code execution, and a local knowledge base."
    ),
    tools=[
        code_interpreter_tool(),
        web_search_tool(),
    ],
)

# Also add custom function tools
@agent.tool
async def query_knowledge_base(topic: str) -> str:
    """Search the internal knowledge base for company-specific information.

    Args:
        topic: The topic to search for in the knowledge base.
    """
    # Simulated KB lookup
    kb = {
        "pricing": "Enterprise: $99/user/mo, Team: $29/user/mo, Free: 5 users",
        "sla": "99.9% uptime, 4hr response for critical, 24hr for standard",
    }
    return kb.get(topic.lower(), f"No knowledge base entry found for '{topic}'")

result = agent.run_sync(
    "What's our current pricing, and how does it compare to the "
    "industry average? Calculate the percentage difference."
)
print(result.output)

4.1 Tool Priority & Selection Behavior

When multiple tools could handle a request, the model selects based on the tool descriptions. Write clear, distinct docstrings to minimize confusion:

from pydantic_ai import Agent

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

@agent.tool
async def search_internal_docs(query: str) -> str:
    """Search INTERNAL company documentation and policies.
    Use this for company-specific questions about processes, policies, or internal systems.

    Args:
        query: The search query for internal docs.
    """
    return f"Internal doc result for: {query}"

@agent.tool
async def search_public_web(query: str) -> str:
    """Search the PUBLIC internet for general knowledge and external information.
    Use this for questions about the outside world, competitors, or general topics.

    Args:
        query: The search query for public web.
    """
    return f"Public web result for: {query}"

# The model picks the right tool based on context
result = agent.run_sync("What's our company's vacation policy?")
print(result.output)  # Uses search_internal_docs

result = agent.run_sync("What's the current weather in Tokyo?")
print(result.output)  # Uses search_public_web

5. Tool Security Considerations

Tools execute real code with real side effects. Every tool is an attack surface — validate inputs rigorously, sandbox execution environments, and rate-limit expensive operations:

5.1 Input Validation & Rate Limiting

from pydantic_ai import Agent, ModelRetry
from pydantic import BaseModel, field_validator
import time
from collections import defaultdict

# Simple rate limiter
class RateLimiter:
    def __init__(self, max_calls: int, period_seconds: float):
        self.max_calls = max_calls
        self.period = period_seconds
        self.calls: list[float] = []

    def check(self) -> bool:
        now = time.time()
        self.calls = [t for t in self.calls if now - t < self.period]
        if len(self.calls) >= self.max_calls:
            return False
        self.calls.append(now)
        return True

# Rate limit: max 10 calls per minute
limiter = RateLimiter(max_calls=10, period_seconds=60)

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

@agent.tool(retries=1)
async def execute_query(sql: str) -> str:
    """Execute a read-only SQL query against the database.

    Args:
        sql: A SELECT query (no modifications allowed).
    """
    # Rate limiting
    if not limiter.check():
        raise ModelRetry("Rate limit exceeded. Please wait before making more queries.")

    # Input validation — block dangerous patterns
    sql_upper = sql.upper().strip()
    blocked_keywords = ["DROP", "DELETE", "INSERT", "UPDATE", "ALTER", "TRUNCATE"]
    for keyword in blocked_keywords:
        if keyword in sql_upper:
            raise ModelRetry(
                f"Query contains blocked keyword '{keyword}'. "
                "Only SELECT queries are allowed."
            )

    if not sql_upper.startswith("SELECT"):
        raise ModelRetry("Query must start with SELECT.")

    # Simulated execution
    return f"Query executed: {sql[:80]}... (3 rows returned)"

result = agent.run_sync("Show me the top 5 customers by revenue")
print(result.output)