Tool Integration Architecture

1# ❌ Bad: Tool does too many things
2class FileOperationsTool:
3    def execute(self, operation: str, path: str, content: str = None):
4        if operation == "read":
5            return open(path).read()
6        elif operation == "write":
7            open(path, "w").write(content)
8        elif operation == "delete":
9            os.remove(path)
10        # ... 10 more operations
11
12# ✅ Good: Separate tools for each operation
13class ReadFileTool:
14    name = "read_file"
15    def execute(self, path: str) -> str:
16        return open(path).read()
17
18class WriteFileTool:
19    name = "write_file"
20    def execute(self, path: str, content: str) -> str:
21        open(path, "w").write(content)
22        return f"Wrote {len(content)} bytes to {path}"

1# ❌ Vague description
2class SearchTool:
3    name = "search"
4    description = "Search for things"
5
6# ✅ Clear, specific description
7class WebSearchTool:
8    name = "web_search"
9    description = """
10Search the web using a search engine and return top results.
11
12Use this tool when you need to:
13- Find current information not in your training data
14- Look up documentation or tutorials
15- Research topics or gather facts
16
17Returns: A list of search results with titles, URLs, and snippets.
18"""

1read_file_schema = {
2    "name": "read_file",
3    "description": "Read the contents of a file",
4    "input_schema": {
5        "type": "object",
6        "properties": {
7            "path": {
8                "type": "string",
9                "description": "Absolute path to the file to read",
10            },
11            "encoding": {
12                "type": "string",
13                "description": "Character encoding (default: utf-8)",
14                "default": "utf-8",
15                "enum": ["utf-8", "ascii", "latin-1"],
16            },
17            "max_lines": {
18                "type": "integer",
19                "description": "Maximum lines to read (default: all)",
20                "minimum": 1,
21            },
22        },
23        "required": ["path"],
24    },
25}

1# ❌ Minimal result
2def execute(self, query: str) -> str:
3    results = search(query)
4    return str(results)
5
6# ✅ Informative result
7def execute(self, query: str) -> str:
8    results = search(query)
9    return f"""
10Search query: "{query}"
11Found {len(results)} results.
12
13Top results:
14{self._format_results(results[:5])}
15
16To see more results, refine your query or increase num_results.
17"""

1class ReadFileTool(Tool):
2    name = "read_file"
3    description = "Read the contents of a file"
4
5    def execute(self, path: str, max_lines: int | None = None) -> ToolResult:
6        try:
7            with open(path, "r") as f:
8                if max_lines:
9                    content = "".join(f.readlines()[:max_lines])
10                else:
11                    content = f.read()
12            return ToolResult(success=True, output=content)
13        except FileNotFoundError:
14            return ToolResult(
15                success=False,
16                output="",
17                error=f"File not found: {path}",
18            )
19
20
21class SearchWebTool(Tool):
22    name = "search_web"
23    description = "Search the web for information"
24
25    def execute(self, query: str, num_results: int = 5) -> ToolResult:
26        try:
27            results = self.search_api.search(query, limit=num_results)
28            formatted = self._format_results(results)
29            return ToolResult(success=True, output=formatted)
30        except Exception as e:
31            return ToolResult(
32                success=False,
33                output="",
34                error=f"Search failed: {e}",
35            )

1class WriteFileTool(Tool):
2    name = "write_file"
3    description = "Write content to a file, creating it if necessary"
4
5    def execute(self, path: str, content: str) -> ToolResult:
6        try:
7            # Create parent directories if needed
8            Path(path).parent.mkdir(parents=True, exist_ok=True)
9
10            with open(path, "w") as f:
11                f.write(content)
12
13            return ToolResult(
14                success=True,
15                output=f"Successfully wrote {len(content)} bytes to {path}",
16            )
17        except PermissionError:
18            return ToolResult(
19                success=False,
20                output="",
21                error=f"Permission denied: {path}",
22            )

1class RunPythonTool(Tool):
2    name = "run_python"
3    description = "Execute Python code and return the output"
4
5    def execute(self, code: str, timeout: int = 30) -> ToolResult:
6        try:
7            # Run in subprocess for isolation
8            result = subprocess.run(
9                ["python", "-c", code],
10                capture_output=True,
11                text=True,
12                timeout=timeout,
13            )
14
15            output = result.stdout
16            if result.stderr:
17                output += f"\nStderr:\n{result.stderr}"
18
19            return ToolResult(
20                success=result.returncode == 0,
21                output=output,
22                error=result.stderr if result.returncode != 0 else None,
23            )
24        except subprocess.TimeoutExpired:
25            return ToolResult(
26                success=False,
27                output="",
28                error=f"Execution timed out after {timeout} seconds",
29            )

1from abc import ABC, abstractmethod
2from dataclasses import dataclass
3from typing import Any
4
5
6@dataclass
7class ToolResult:
8    """Standardized result from tool execution."""
9    success: bool
10    output: str
11    error: str | None = None
12    metadata: dict | None = None
13
14
15class Tool(ABC):
16    """Abstract base class for all tools."""
17
18    name: str
19    description: str
20    dangerous: bool = False  # Requires confirmation
21
22    @abstractmethod
23    def get_schema(self) -> dict:
24        """Return the JSON schema for LLM function calling."""
25        pass
26
27    @abstractmethod
28    def execute(self, **kwargs: Any) -> ToolResult:
29        """Execute the tool with given parameters."""
30        pass
31
32    def validate_params(self, **kwargs: Any) -> tuple[bool, str | None]:
33        """Validate parameters before execution."""
34        return True, None

1from typing import get_type_hints
2import inspect
3
4
5def generate_schema(tool: Tool) -> dict:
6    """Generate JSON schema from tool execute method signature."""
7
8    sig = inspect.signature(tool.execute)
9    hints = get_type_hints(tool.execute)
10
11    properties = {}
12    required = []
13
14    for param_name, param in sig.parameters.items():
15        if param_name in ("self", "kwargs"):
16            continue
17
18        param_type = hints.get(param_name, str)
19        json_type = python_to_json_type(param_type)
20
21        properties[param_name] = {
22            "type": json_type,
23            "description": get_param_description(tool, param_name),
24        }
25
26        if param.default is inspect.Parameter.empty:
27            required.append(param_name)
28
29    return {
30        "name": tool.name,
31        "description": tool.description,
32        "input_schema": {
33            "type": "object",
34            "properties": properties,
35            "required": required,
36        },
37    }

1class ToolRegistry:
2    """Centralized registry for agent tools."""
3
4    def __init__(self):
5        self._tools: dict[str, Tool] = {}
6
7    def register(self, tool: Tool) -> "ToolRegistry":
8        """Register a tool. Returns self for chaining."""
9        self._tools[tool.name] = tool
10        return self
11
12    def register_many(self, tools: list[Tool]) -> "ToolRegistry":
13        """Register multiple tools at once."""
14        for tool in tools:
15            self.register(tool)
16        return self
17
18    def get(self, name: str) -> Tool | None:
19        """Get a tool by name."""
20        return self._tools.get(name)
21
22    def list_tools(self) -> list[str]:
23        """List all registered tool names."""
24        return list(self._tools.keys())
25
26    def get_schemas(self) -> list[dict]:
27        """Get schemas for all tools (for LLM)."""
28        return [tool.get_schema() for tool in self._tools.values()]
29
30    def execute(self, name: str, **kwargs) -> ToolResult:
31        """Execute a tool by name."""
32        tool = self.get(name)
33        if not tool:
34            return ToolResult(
35                success=False,
36                output="",
37                error=f"Unknown tool: {name}",
38            )
39
40        # Validate parameters
41        valid, error = tool.validate_params(**kwargs)
42        if not valid:
43            return ToolResult(
44                success=False,
45                output="",
46                error=f"Invalid parameters: {error}",
47            )
48
49        return tool.execute(**kwargs)
50
51
52# Usage
53registry = ToolRegistry()
54registry.register_many([
55    ReadFileTool(),
56    WriteFileTool(),
57    SearchWebTool(),
58    RunPythonTool(),
59])

1class Tool(ABC):
2    def safe_execute(self, **kwargs) -> ToolResult:
3        """Execute with comprehensive error handling."""
4        try:
5            # Validate parameters
6            valid, error = self.validate_params(**kwargs)
7            if not valid:
8                return ToolResult(
9                    success=False,
10                    output="",
11                    error=f"Parameter validation failed: {error}",
12                )
13
14            # Execute with timeout
15            result = self._execute_with_timeout(**kwargs)
16            return result
17
18        except TimeoutError:
19            return ToolResult(
20                success=False,
21                output="",
22                error="Tool execution timed out",
23            )
24        except PermissionError as e:
25            return ToolResult(
26                success=False,
27                output="",
28                error=f"Permission denied: {e}",
29            )
30        except Exception as e:
31            # Log unexpected errors
32            logger.exception(f"Tool {self.name} failed unexpectedly")
33            return ToolResult(
34                success=False,
35                output="",
36                error=f"Unexpected error: {type(e).__name__}: {e}",
37            )