Building AI Agent Tools: Custom Tool Development Guide

from typing import List, Dict, Any, Optional
from pydantic import BaseModel

class ToolParameter(BaseModel):
    """Single parameter definition"""
    name: str
    type: str  # "string", "integer", "boolean", "object", "array"
    description: str
    required: bool = False
    default: Optional[Any] = None
    enum: Optional[List[str]] = None

class Tool(BaseModel):
    """Complete tool definition"""
    name: str
    description: str
    parameters: List[ToolParameter]
    category: Optional[str] = None
    tags: List[str] = []
    examples: Optional[List[Dict]] = None
    
    def to_openai_format(self) -> Dict:
        """Convert to OpenAI function calling format"""
        return {
            "type": "function",
            "function": {
                "name": self.name,
                "description": self.description,
                "parameters": {
                    "type": "object",
                    "properties": {
                        param.name: {
                            "type": param.type,
                            "description": param.description,
                            **({"enum": param.enum} if param.enum else {}),
                            **({"default": param.default} if param.default else {})
                        }
                        for param in self.parameters
                    },
                    "required": [p.name for p in self.parameters if p.required]
                }
            }
        }

import requests
from typing import Optional

class WebSearchTool:
    """Tool for searching the web"""
    
    name = "web_search"
    description = "Search the web for current information"
    
    parameters = [
        ToolParameter(
            name="query",
            type="string",
            description="The search query",
            required=True
        ),
        ToolParameter(
            name="num_results",
            type="integer",
            description="Number of results to return",
            required=False,
            default=5
        )
    ]
    
    def __init__(self, api_key: str):
        self.api_key = api_key
    
    async def execute(self, query: str, num_results: int = 5) -> Dict:
        """Execute the search"""
        
        # Using DuckDuckGo API (free) or Google/Bing
        url = "https://api.duckduckgo.com/"
        params = {
            "q": query,
            "format": "json",
            "no_html": 1,
            "skip_disambig": 1
        }
        
        response = requests.get(url, params=params)
        data = response.json()
        
        results = []
        for item in data.get("RelatedTopics", [])[:num_results]:
            results.append({
                "title": item.get("Text", ""),
                "url": item.get("URL", ""),
                "snippet": item.get("Text", "")[:200]
            })
        
        return {
            "query": query,
            "results": results,
            "count": len(results)
        }

import ast
import operator

class CalculatorTool:
    """Safe mathematical expression evaluator"""
    
    name = "calculate"
    description = "Evaluate mathematical expressions"
    
    parameters = [
        ToolParameter(
            name="expression",
            type="string",
            description="Mathematical expression to evaluate",
            required=True
        )
    ]
    
    # Safe operators
    SAFE_OPERATORS = {
        ast.Add: operator.add,
        ast.Sub: operator.sub,
        ast.Mult: operator.mul,
        ast.Div: operator.truediv,
        ast.Pow: operator.pow,
        ast.USub: operator.neg,
    }
    
    def execute(self, expression: str) -> Dict:
        """Safely evaluate mathematical expression"""
        
        try:
            # Parse expression
            tree = ast.parse(expression, mode='eval')
            
            # Evaluate safely
            result = self._eval_node(tree.body)
            
            return {
                "expression": expression,
                "result": result,
                "success": True
            }
            
        except Exception as e:
            return {
                "expression": expression,
                "result": None,
                "success": False,
                "error": str(e)
            }
    
    def _eval_node(self, node):
        if isinstance(node, ast.Num):
            return node.n
        elif isinstance(node, ast.BinOp):
            left = self._eval_node(node.left)
            right = self._eval_node(node.right)
            return self.SAFE_OPERATORS[type(node.op)](left, right)
        elif isinstance(node, ast.UnaryOp):
            operand = self._eval_node(node.operand)
            return self.SAFE_OPERATORS[type(node.op)](operand)
        else:
            raise ValueError(f"Unsupported operation: {type(node)}")

from pathlib import Path
import os

class FileOperationsTool:
    """Tool for file operations"""
    
    name = "file_operations"
    description = "Read, write, and manage files"
    
    parameters = [
        ToolParameter(
            name="operation",
            type="string",
            description="Operation: read, write, list, delete",
            required=True,
            enum=["read", "write", "list", "delete", "exists"]
        ),
        ToolParameter(
            name="path",
            type="string",
            description="File or directory path",
            required=True
        ),
        ToolParameter(
            name="content",
            type="string",
            description="Content to write (for write operation)",
            required=False
        )
    ]
    
    def __init__(self, allowed_directories: List[str]):
        self.allowed_dirs = allowed_directories
    
    def _validate_path(self, path: str) -> bool:
        """Ensure path is within allowed directories"""
        abs_path = Path(path).resolve()
        return any(
            str(abs_path).startswith(dir) 
            for dir in self.allowed_dirs
        )
    
    def execute(self, operation: str, path: str, content: str = None) -> Dict:
        """Execute file operation"""
        
        if not self._validate_path(path):
            return {"success": False, "error": "Path not allowed"}
        
        try:
            if operation == "read":
                p = Path(path)
                if not p.exists():
                    return {"success": False, "error": "File not found"}
                return {"success": True, "content": p.read_text()}
            
            elif operation == "write":
                p = Path(path)
                p.parent.mkdir(parents=True, exist_ok=True)
                p.write_text(content or "")
                return {"success": True, "path": str(p)}
            
            elif operation == "list":
                p = Path(path)
                if not p.is_dir():
                    return {"success": False, "error": "Not a directory"}
                return {
                    "success": True,
                    "files": [f.name for f in p.iterdir()]
                }
            
            elif operation == "delete":
                p = Path(path)
                if p.is_file():
                    p.unlink()
                elif p.is_dir():
                    p.rmdir()
                return {"success": True}
            
            elif operation == "exists":
                return {"success": True, "exists": Path(path).exists()}
        
        except Exception as e:
            return {"success": False, "error": str(e)}

class ToolRegistry:
    """Central registry for all agent tools"""
    
    def __init__(self):
        self.tools: Dict[str, Tool] = {}
        self.categories: Dict[str, List[str]] = {}
    
    def register(self, tool: Tool):
        """Register a tool"""
        self.tools[tool.name] = tool
        
        # Add to category
        if tool.category:
            if tool.category not in self.categories:
                self.categories[tool.category] = []
            self.categories[tool.category].append(tool.name)
    
    def get(self, name: str) -> Optional[Tool]:
        """Get tool by name"""
        return self.tools.get(name)
    
    def get_all(self) -> List[Tool]:
        """Get all tools"""
        return list(self.tools.values())
    
    def get_by_category(self, category: str) -> List[Tool]:
        """Get tools in a category"""
        tool_names = self.categories.get(category, [])
        return [self.tools[name] for name in tool_names]
    
    def get_schema(self) -> List[Dict]:
        """Get OpenAI-format tool schema"""
        return [tool.to_openai_format() for tool in self.tools.values()]
    
    def search(self, query: str) -> List[Tool]:
        """Search tools by name or description"""
        query = query.lower()
        return [
            tool for tool in self.tools.values()
            if query in tool.name.lower() or query in tool.description.lower()
        ]

# Usage
registry = ToolRegistry()
registry.register(WebSearchTool(api_key="xxx"))
registry.register(CalculatorTool())
registry.register(FileOperationsTool(allowed_directories=["./workspace"]))

class DynamicToolLoader:
    """Load tools dynamically from modules"""
    
    def __init__(self, registry: ToolRegistry):
        self.registry = registry
    
    def load_from_directory(self, directory: str):
        """Load all tools from a directory"""
        import importlib.util
        from pathlib import Path
        
        for file in Path(directory).glob("*.py"):
            if file.stem.startswith("_"):
                continue
            
            # Import module
            spec = importlib.util.spec_from_file_location(
                file.stem, 
                file
            )
            module = importlib.util.module_from_spec(spec)
            spec.loader.exec_module(module)
            
            # Find Tool classes
            for name in dir(module):
                obj = getattr(module, name)
                if (isinstance(obj, type) and 
                    issubclass(obj, BaseTool) and 
                    obj != BaseTool):
                    
                    # Instantiate and register
                    tool = obj()
                    self.registry.register(tool)
    
    def load_from_package(self, package_name: str):
        """Load tools from a Python package"""
        import importlib
        package = importlib.import_module(package_name)
        
        for name in dir(package):
            obj = getattr(package, name)
            if (isinstance(obj, type) and 
                issubclass(obj, BaseTool) and 
                obj != BaseTool):
                
                tool = obj()
                self.registry.register(tool)

from mcp.types import Tool as MCPTool
from mcp import MCPServer

class MCPToolAdapter:
    """Adapter to convert tools to MCP format"""
    
    @staticmethod
    def to_mcp_tool(tool: Tool) -> MCPTool:
        """Convert custom tool to MCP tool"""
        
        return MCPTool(
            name=tool.name,
            description=tool.description,
            inputSchema={
                "type": "object",
                "properties": {
                    param.name: {
                        "type": param.type,
                        "description": param.description
                    }
                    for param in tool.parameters
                },
                "required": [p.name for p in tool.parameters if p.required]
            }
        )
    
    @staticmethod
    def from_mcp_tool(mcp_tool: MCPTool) -> Tool:
        """Convert MCP tool to custom tool"""
        
        params = []
        for name, schema in mcp_tool.inputSchema.get("properties", {}).items():
            params.append(ToolParameter(
                name=name,
                type=schema.get("type", "string"),
                description=schema.get("description", ""),
                required=name in mcp_tool.inputSchema.get("required", [])
            ))
        
        return Tool(
            name=mcp_tool.name,
            description=mcp_tool.description,
            parameters=params
        )

class MyMCPServer(MCPServer):
    """Custom MCP server with tools"""
    
    def __init__(self):
        super().__init__()
        self.registry = ToolRegistry()
        self._register_tools()
    
    def _register_tools(self):
        # Register custom tools
        self.registry.register(MyCustomTool())
    
    async def list_tools(self) -> List[MCPTool]:
        """List available tools"""
        return [
            MCPToolAdapter.to_mcp_tool(tool) 
            for tool in self.registry.get_all()
        ]
    
    async def call_tool(self, name: str, arguments: Dict) -> Any:
        """Call a tool"""
        tool = self.registry.get(name)
        if not tool:
            raise ValueError(f"Unknown tool: {name}")
        
        return await tool.execute(**arguments)

class ToolChain:
    """Chain multiple tools together"""
    
    def __init__(self, registry: ToolRegistry):
        self.registry = registry
    
    async def execute_chain(
        self, 
        chain_definition: List[Dict]
    ) -> List[Dict]:
        """Execute a chain of tools"""
        
        results = []
        context = {}
        
        for step in chain_definition:
            tool_name = step["tool"]
            input_mapping = step.get("input", {})
            
            # Build input from context and mapping
            tool_input = {}
            for key, value in input_mapping.items():
                if value.startswith("$"):
                    # Reference previous result
                    var = value[1:]
                    tool_input[key] = context.get(var)
                else:
                    tool_input[key] = value
            
            # Execute tool
            tool = self.registry.get(tool_name)
            result = await tool.execute(**tool_input)
            
            # Store in context
            context[step.get("output_var", "result")] = result
            results.append(result)
        
        return results


# Example chain
chain = [
    {
        "tool": "web_search",
        "input": {"query": "latest AI developments"},
        "output_var": "search_results"
    },
    {
        "tool": "summarize",
        "input": {"text": "$search_results"},
        "output_var": "summary"
    },
    {
        "tool": "save_to_file",
        "input": {"content": "$summary", "path": "ai_news.md"}
    }
]

class ToolWithFallbacks:
    """Tool with fallback options"""
    
    def __init__(self):
        self.primary = PrimarySearchTool()
        self.fallbacks = [BackupSearchTool(), DuckDuckGoTool()]
    
    async def execute(self, query: str) -> Dict:
        """Try primary, fallback on failure"""
        
        try:
            return await self.primary.execute(query)
        except Exception as e:
            # Try fallbacks
            for fallback in self.fallbacks:
                try:
                    result = await fallback.execute(query)
                    result["via_fallback"] = fallback.name
                    return result
                except:
                    continue
            
            return {
                "success": False,
                "error": f"All tools failed: {e}"
            }

import hashlib
import json
from datetime import timedelta

class CachedTool:
    """Tool with result caching"""
    
    def __init__(self, tool: Tool, cache_ttl: timedelta = timedelta(hours=1)):
        self.tool = tool
        self.cache = {}
        self.cache_ttl = cache_ttl
    
    def _get_cache_key(self, **kwargs) -> str:
        """Generate cache key"""
        key_data = json.dumps(kwargs, sort_keys=True)
        return hashlib.md5(key_data.encode()).hexdigest()
    
    async def execute(self, **kwargs) -> Dict:
        """Execute with caching"""
        
        cache_key = self._get_cache_key(**kwargs)
        
        # Check cache
        if cache_key in self.cache:
            cached = self.cache[cache_key]
            if datetime.now() < cached["expires"]:
                cached["result"]["cached"] = True
                return cached["result"]
        
        # Execute
        result = await self.tool.execute(**kwargs)
        
        # Cache result
        self.cache[cache_key] = {
            "result": result,
            "expires": datetime.now() + self.cache_ttl
        }
        
        return result

# Good: Clear, specific descriptions
class GoodSearchTool:
    name = "search_codebase"
    description = """
    Search the codebase for specific code patterns or files.
    Use this to find function definitions, imports, or code patterns.
    Supports regex patterns and file type filtering.
    """  # Clear purpose
    
    parameters = [
        ToolParameter(
            name="pattern",
            type="string",
            description="Search pattern (supports regex)"  # Specific
        ),
        ToolParameter(
            name="file_type",
            type="string", 
            description="Filter by file extension (e.g., 'py', 'js')",
            required=False
        )
    ]

# Bad: Unclear, generic descriptions
class BadSearchTool:
    name = "search"
    description = "Search for things"  # Too vague
    
    parameters = [
        ToolParameter(name="q", type="string", description="Query")  # Unclear
    ]

class RobustTool:
    async def execute(self, **kwargs) -> Dict:
        try:
            result = await self._do_execution(**kwargs)
            return {
                "success": True,
                "data": result
            }
        except ValidationError as e:
            return {
                "success": False,
                "error": "Invalid input",
                "details": str(e)
            }
        except PermissionError:
            return {
                "success": False,
                "error": "Permission denied"
            }
        except Exception as e:
            return {
                "success": False,
                "error": "Execution failed",
                "details": str(e)
            }