# Custom Tools

Building, registering, and managing custom tools for agents.

---

## Tool Anatomy

### Tool Interface
```python
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
```python
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
```python
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
```python
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
```python
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
```python
# pyproject.toml
[project.entry-points."myapp.tools"]
web_search = "myapp.tools.web:WebSearchTool"
code_exec = "myapp.tools.code:CodeExecutionTool"
custom_tool = "myplugin.tools:CustomTool"
```

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

---

## Tool Composition

### Chaining Tools
```python
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
```python
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
```python
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
```python
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
```python
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
```python
@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
```python
@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
```python
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
```python
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
```python
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
```python
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
```

---

## Related Notes

- [[ToolUse/Function-Calling.md]] — Tools as functions
- [[ToolUse/MCP-Protocol.md]] — Tools via MCP
- [[ToolUse/Code-Execution.md]] — Code execution tools
- [[ToolUse/Browser-Automation.md]] — Browser tools
- [[ToolUse/API-Integration.md]] — API tools
- [[ToolUse/Tool-Discovery.md]] — Dynamic discovery
- [[ToolUse/Tool-Discovery.md]] — Dynamic discovery
- [[Frameworks/OpenAI-Agents-SDK.md]] — Tools in OpenAI SDK
- [[Frameworks/LangGraph.md]] — Tools in LangGraph
- [[Frameworks/PydanticAI.md]] — Type-safe tools
- [[Frameworks/LangGraph.md]] — Tools in LangGraph
- [[Architectures/Single-Agent-Patterns.md]] — ReAct, tool use
- [[Architectures/Control-Flow.md]] — Tool control flow
- [[Production/Observability.md]] — Tool observability
- [[Production/Security.md]] — Tool security
- [[Security/Secure-Coding.md]] — 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.*