Custom Tools

Building, registering, and managing custom tools for agents.

Tool Anatomy

Tool Interface

from abc import ABC, abstractmethod
from pydantic import BaseModel
from typing import Any

class ToolInput(BaseModel):
    """Input schema - defines what the tool accepts"""
    pass

class ToolOutput(BaseModel):
    """Output schema - defines what the tool returns"""
    pass

class Tool(ABC):
    name: str
    description: str
    input_schema: type[ToolInput]
    output_schema: type[ToolOutput] | None = None

    # Metadata
    idempotent: bool = False
    side_effects: list[str] = []  # "write", "delete", "cost", "external", "network"
    timeout_seconds: float = 30.0
    retryable: bool = True

    @abstractmethod
    async def execute(self, input: ToolInput) -> ToolOutput:
        pass

    def validate_input(self, data: dict) -> ToolInput:
        return self.input_schema(**data)

    def validate_output(self, data: Any) -> ToolOutput:
        if self.output_schema:
            return self.output_schema(**data)
        return data

Example: Complete Tool

from pydantic import BaseModel, Field, HttpUrl
from typing import Literal

class SearchInput(BaseModel):
    query: str = Field(..., description="Search query, optimized for search engines")
    max_results: int = Field(10, ge=1, le=50, description="Maximum results to return")
    recency_days: int | None = Field(None, ge=1, le=365, description="Limit to last N days")
    safe_search: bool = Field(True, description="Enable safe search filter")

class SearchResult(BaseModel):
    title: str
    url: HttpUrl
    snippet: str
    published_date: str | None = None

class SearchOutput(BaseModel):
    results: list[SearchResult]
    total_results: int
    search_time_ms: int

class WebSearchTool(Tool):
    name = "web_search"
    description = "Search the web for current information. Use for recent events, facts, data, documentation."
    input_schema = SearchInput
    output_schema = SearchOutput
    idempotent = True
    side_effects = ["network", "cost"]
    timeout_seconds = 15.0

    def __init__(self, search_provider: SearchProvider):
        self.provider = search_provider

    async def execute(self, input: SearchInput) -> SearchOutput:
        start = time.time()
        results = await self.provider.search(
            query=input.query,
            max_results=input.max_results,
            recency_days=input.recency_days,
            safe_search=input.safe_search
        )
        return SearchOutput(
            results=results,
            total_results=len(results),
            search_time_ms=int((time.time() - start) * 1000)
        )

Tool Registration

Registry Pattern

class ToolRegistry:
    def __init__(self):
        self._tools: dict[str, Tool] = {}
        self._categories: dict[str, set[str]] = defaultdict(set)

    def register(self, tool: Tool, category: str = "general") -> None:
        if tool.name in self._tools:
            raise ValueError(f"Tool '{tool.name}' already registered")
        self._tools[tool.name] = tool
        self._categories[category].add(tool.name)

    def get(self, name: str) -> Tool | None:
        return self._tools.get(name)

    def list_tools(self, category: str | None = None) -> list[Tool]:
        if category:
            return [self._tools[name] for name in self._categories.get(category, set())]
        return list(self._tools.values())

    def get_schemas(self) -> list[dict]:
        """Return OpenAI-compatible function schemas"""
        return [
            {
                "type": "function",
                "function": {
                    "name": tool.name,
                    "description": tool.description,
                    "parameters": tool.input_schema.model_json_schema()
                }
            }
            for tool in self._tools.values()
        ]

# Global registry
registry = ToolRegistry()

Auto-Registration via Decorator

def tool(category: str = "general"):
    def decorator(cls: type[Tool]):
        instance = cls()
        registry.register(instance, category)
        return cls
    return decorator

@tool("web")
class WebSearchTool(Tool):
    ...

@tool("code")
class CodeExecutionTool(Tool):
    ...

Tool Discovery

Dynamic Loading

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

    def discover_from_module(self, module_name: str) -> list[Tool]:
        """Import module and register any @tool decorated classes"""
        module = importlib.import_module(module_name)
        tools = []
        for name in dir(module):
            obj = getattr(module, name)
            if isinstance(obj, Tool):
                self.registry.register(obj)
                tools.append(obj)
        return tools

    def discover_from_package(self, package_name: str) -> list[Tool]:
        """Discover all tools in a package"""
        package = importlib.import_module(package_name)
        tools = []
        for _, module_name, _ in pkgutil.iter_modules(package.__path__):
            full_name = f"{package_name}.{module_name}"
            tools.extend(self.discover_from_module(full_name))
        return tools

    def discover_from_entry_points(self, group: str = "myapp.tools") -> list[Tool]:
        """Discover tools via setuptools entry points"""
        tools = []
        for entry_point in importlib.metadata.entry_points(group=group):
            tool_class = entry_point.load()
            tool_instance = tool_class()
            self.registry.register(tool_instance)
            tools.append(tool_instance)
        return tools

Plugin System

# pyproject.toml
[project.entry-points."myapp.tools"]
web_search = "myapp.tools.web:WebSearchTool"
code_exec = "myapp.tools.code:CodeExecutionTool"
custom_tool = "myplugin.tools:CustomTool"

# Load all plugins
discovery = ToolDiscovery(registry)
discovery.discover_from_entry_points("myapp.tools")

Tool Composition

Chaining Tools

class ToolChain:
    def __init__(self, tools: list[Tool]):
        self.tools = tools

    async def execute(self, initial_input: dict) -> dict:
        context = initial_input
        for tool in self.tools:
            # Extract relevant input for this tool
            tool_input = self._extract_input(tool, context)
            result = await tool.execute(tool.validate_input(tool_input))
            context = {**context, tool.name: result}
        return context

Parallel Execution

class ParallelToolExecutor:
    def __init__(self, max_concurrent: int = 5):
        self.semaphore = asyncio.Semaphore(max_concurrent)

    async def execute_parallel(self, tool_calls: list[tuple[Tool, dict]]) -> list[Any]:
        async def execute_one(tool: Tool, input_data: dict):
            async with self.semaphore:
                validated = tool.validate_input(input_data)
                return await tool.execute(validated)

        tasks = [
            execute_one(tool, input_data) 
            for tool, input_data in tool_calls
        ]
        return await asyncio.gather(*tasks, return_exceptions=True)

Conditional Execution

class ConditionalToolExecutor:
    def __init__(self):
        self.rules: list[tuple[Callable[[dict], bool], Tool]] = []

    def add_rule(self, condition: Callable[[dict], bool], tool: Tool):
        self.rules.append((condition, tool))

    async def execute(self, context: dict) -> dict:
        for condition, tool in self.rules:
            if condition(context):
                result = await tool.execute(tool.validate_input(context))
                context[tool.name] = result
        return context

Tool Versioning

Versioned Tools

class VersionedTool(Tool):
    version: str = "1.0.0"
    deprecated: bool = False
    replacement: str | None = None  # Name of replacement tool

    def get_schema(self) -> dict:
        schema = super().get_schema()
        schema["function"]["version"] = self.version
        if self.deprecated:
            schema["function"]["deprecated"] = True
            if self.replacement:
                schema["function"]["replacement"] = self.replacement
        return schema

# Usage
class WebSearchToolV1(VersionedTool):
    version = "1.0.0"
    deprecated = True
    replacement = "web_search_v2"
    ...

class WebSearchToolV2(VersionedTool):
    version = "2.0.0"
    ...

Version Resolution

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

    def resolve(self, tool_name: str, version: str | None = None) -> Tool:
        if version:
            # Exact version match
            return self.registry.get(f"{tool_name}_v{version}")

        # Get latest non-deprecated
        candidates = [
            t for t in self.registry.list_tools() 
            if t.name.startswith(f"{tool_name}_v") and not t.deprecated
        ]
        if not candidates:
            # Fallback to any version
            candidates = [t for t in self.registry.list_tools() if t.name.startswith(f"{tool_name}_v")]

        return max(candidates, key=lambda t: tuple(map(int, t.version.split('.'))))

Tool Testing

Unit Testing

@pytest.fixture
def search_tool():
    return WebSearchTool(mock_search_provider())

async def test_web_search_tool(search_tool):
    # Test valid input
    input_data = SearchInput(query="Python tutorial", max_results=5)
    result = await search_tool.execute(input_data)

    assert isinstance(result, SearchOutput)
    assert len(result.results) <= 5
    assert result.search_time_ms > 0

    # Test validation
    with pytest.raises(ValidationError):
        search_tool.validate_input({"query": "", "max_results": 100})  # Invalid

async def test_tool_timeout(search_tool):
    with pytest.raises(TimeoutError):
        await asyncio.wait_for(
            search_tool.execute(SearchInput(query="test", max_results=1000)),
            timeout=0.001
        )

Integration Testing

@pytest.mark.integration
async def test_real_web_search():
    tool = WebSearchTool(RealSearchProvider())
    result = await tool.execute(SearchInput(query="Python", max_results=3))

    assert len(result.results) == 3
    for r in result.results:
        assert r.title
        assert r.url
        assert r.snippet

Property-Based Testing

from hypothesis import given, strategies as st

@given(
    query=st.text(min_size=1, max_size=100),
    max_results=st.integers(min_value=1, max_value=50)
)
async def test_search_input_validation(query, max_results):
    input_data = SearchInput(query=query, max_results=max_results)
    # Should not raise
    validated = WebSearchTool.input_schema(**input_data.model_dump())
    assert validated.query == query
    assert validated.max_results == max_results

Tool Observability

Logging

class ObservableTool(Tool):
    def __init__(self, tool: Tool, logger: Logger):
        self.tool = tool
        self.logger = logger
        # Delegate attributes
        self.name = tool.name
        self.description = tool.description
        self.input_schema = tool.input_schema
        self.output_schema = tool.output_schema

    async def execute(self, input: ToolInput) -> ToolOutput:
        start = time.time()
        call_id = str(uuid4())

        self.logger.info("Tool call started", extra={
            "tool": self.tool.name,
            "call_id": call_id,
            "input": input.model_dump()
        })

        try:
            result = await self.tool.execute(input)
            duration = time.time() - start

            self.logger.info("Tool call completed", extra={
                "tool": self.tool.name,
                "call_id": call_id,
                "duration_ms": duration * 1000,
                "success": True
            })
            return result
        except Exception as e:
            duration = time.time() - start
            self.logger.error("Tool call failed", extra={
                "tool": self.tool.name,
                "call_id": call_id,
                "duration_ms": duration * 1000,
                "error": str(e)
            })
            raise

Metrics

from prometheus_client import Histogram, Counter

TOOL_CALL_DURATION = Histogram(
    "tool_call_duration_seconds",
    "Tool call duration",
    ["tool", "success"]
)

TOOL_CALLS_TOTAL = Counter(
    "tool_calls_total",
    "Total tool calls",
    ["tool", "success"]
)

class MetricsTool(Tool):
    def __init__(self, tool: Tool):
        self.tool = tool
        # Delegate...

    async def execute(self, input: ToolInput) -> ToolOutput:
        start = time.time()
        success = True
        try:
            return await self.tool.execute(input)
        except Exception:
            success = False
            raise
        finally:
            duration = time.time() - start
            TOOL_CALL_DURATION.labels(self.tool.name, str(success)).observe(time.time() - start)
            TOOL_CALLS_TOTAL.labels(self.tool.name, str(success)).inc()

Tool Lifecycle Management

Lifecycle Hooks

class LifecycleTool(Tool):
    async def initialize(self) -> None:
        """Called when tool is registered/loaded"""
        pass

    async def shutdown(self) -> None:
        """Called when tool is unregistered/shutdown"""
        pass

    async def health_check(self) -> HealthStatus:
        """Periodic health check"""
        return HealthStatus.HEALTHY

class ToolLifecycleManager:
    def __init__(self, registry: ToolRegistry):
        self.registry = registry
        self.initialized: set[str] = set()

    async def initialize_all(self):
        for tool in self.registry.list_tools():
            if hasattr(tool, 'initialize') and tool.name not in self.initialized:
                await tool.initialize()
                self.initialized.add(tool.name)

    async def shutdown_all(self):
        for tool in self.registry.list_tools():
            if hasattr(tool, 'shutdown'):
                await tool.shutdown()

    async def health_check_all(self) -> dict[str, HealthStatus]:
        results = {}
        for tool in self.registry.list_tools():
            if hasattr(tool, 'health_check'):
                results[tool.name] = await tool.health_check()
        return results

Function Calling — Tools as functions
MCP Protocol — Tools via MCP
Code Execution — Code execution tools
Browser Automation — Browser tools
API Integration — API tools
Tool Discovery — Dynamic discovery
Tool Discovery — Dynamic discovery
OpenAI Agents SDK — Tools in OpenAI SDK
LangGraph — Tools in LangGraph
PydanticAI — Type-safe tools
LangGraph — Tools in LangGraph
Single Agent Patterns — ReAct, tool use
Control Flow — Tool control flow
Observability — Tool observability
Security — Tool security
Secure Coding — Tool security

Sources: OpenAI function calling, Anthropic tool use, LangChain tools, LangGraph tools, PydanticAI tools, OpenAI Agents SDK tools, MCP spec, plugin architectures, observability patterns.

From the Agentic Engineering research vault · CC BY 4.0 · all notes · raw markdown