Adding Tools

1from dataclasses import dataclass, field
2from typing import Any, Callable, Optional
3import json
4
5@dataclass
6class ToolParameter:
7    """Definition of a tool parameter."""
8    name: str
9    type: str  # "string", "integer", "boolean", "array", "object"
10    description: str
11    required: bool = True
12    default: Any = None
13    enum: Optional[list] = None
14
15@dataclass
16class Tool:
17    """Complete tool definition."""
18    name: str
19    description: str
20    parameters: list[ToolParameter]
21    function: Callable
22    category: str = "general"
23    requires_confirmation: bool = False
24
25    def to_api_schema(self) -> dict:
26        """Convert to Anthropic API format."""
27        properties = {}
28        required = []
29
30        for param in self.parameters:
31            prop = {
32                "type": param.type,
33                "description": param.description
34            }
35            if param.enum:
36                prop["enum"] = param.enum
37            if param.default is not None:
38                prop["default"] = param.default
39
40            properties[param.name] = prop
41
42            if param.required:
43                required.append(param.name)
44
45        return {
46            "name": self.name,
47            "description": self.description,
48            "input_schema": {
49                "type": "object",
50                "properties": properties,
51                "required": required
52            }
53        }
54
55    async def execute(self, **kwargs) -> str:
56        """Execute the tool with given arguments."""
57        import asyncio
58
59        try:
60            if asyncio.iscoroutinefunction(self.function):
61                result = await self.function(**kwargs)
62            else:
63                result = self.function(**kwargs)
64            return str(result)
65        except Exception as e:
66            return f"Error: {str(e)}"

1from functools import wraps
2from typing import get_type_hints
3
4def tool(
5    name: str = None,
6    description: str = None,
7    category: str = "general",
8    requires_confirmation: bool = False
9):
10    """Decorator to create tools from functions."""
11
12    def decorator(fn: Callable) -> Tool:
13        tool_name = name or fn.__name__
14        tool_desc = description or fn.__doc__ or "No description"
15
16        # Extract parameters from type hints and docstring
17        hints = get_type_hints(fn)
18        parameters = []
19
20        for param_name, param_type in hints.items():
21            if param_name == "return":
22                continue
23
24            type_map = {
25                str: "string",
26                int: "integer",
27                float: "number",
28                bool: "boolean",
29                list: "array",
30                dict: "object"
31            }
32
33            parameters.append(ToolParameter(
34                name=param_name,
35                type=type_map.get(param_type, "string"),
36                description=f"The {param_name} parameter",
37                required=True
38            ))
39
40        return Tool(
41            name=tool_name,
42            description=tool_desc,
43            parameters=parameters,
44            function=fn,
45            category=category,
46            requires_confirmation=requires_confirmation
47        )
48
49    return decorator
50
51
52# Usage example
53@tool(
54    name="web_search",
55    description="Search the web for information",
56    category="information"
57)
58def search_web(query: str, num_results: int = 5) -> str:
59    """Search the web and return results."""
60    # Implementation here
61    return f"Search results for: {query}"

1import httpx
2from typing import Optional
3
4class WebSearchTool:
5    """Search the web using a search API."""
6
7    def __init__(self, api_key: str, engine: str = "google"):
8        self.api_key = api_key
9        self.engine = engine
10
11    async def search(
12        self,
13        query: str,
14        num_results: int = 5,
15        site: Optional[str] = None
16    ) -> str:
17        """
18        Search the web.
19
20        Args:
21            query: Search query
22            num_results: Number of results to return
23            site: Limit to specific site (e.g., "python.org")
24
25        Returns:
26            Formatted search results
27        """
28        if site:
29            query = f"site:{site} {query}"
30
31        # Example using SerpAPI (replace with your preferred API)
32        async with httpx.AsyncClient() as client:
33            response = await client.get(
34                "https://serpapi.com/search",
35                params={
36                    "q": query,
37                    "num": num_results,
38                    "api_key": self.api_key
39                }
40            )
41            data = response.json()
42
43        results = []
44        for item in data.get("organic_results", [])[:num_results]:
45            results.append(
46                f"Title: {item['title']}\n"
47                f"URL: {item['link']}\n"
48                f"Snippet: {item.get('snippet', 'No snippet')}"
49            )
50
51        return "\n\n".join(results) if results else "No results found"
52
53    def as_tool(self) -> Tool:
54        """Convert to Tool object."""
55        return Tool(
56            name="web_search",
57            description="Search the web for current information",
58            parameters=[
59                ToolParameter("query", "string", "Search query"),
60                ToolParameter("num_results", "integer", "Number of results", required=False),
61                ToolParameter("site", "string", "Limit to specific site", required=False),
62            ],
63            function=self.search,
64            category="information"
65        )

1class HTTPTool:
2    """Make HTTP requests."""
3
4    def __init__(self, timeout: int = 30):
5        self.timeout = timeout
6
7    async def request(
8        self,
9        url: str,
10        method: str = "GET",
11        headers: dict = None,
12        body: str = None
13    ) -> str:
14        """
15        Make an HTTP request.
16
17        Args:
18            url: The URL to request
19            method: HTTP method (GET, POST, PUT, DELETE)
20            headers: Optional headers
21            body: Optional request body
22
23        Returns:
24            Response text or error
25        """
26        async with httpx.AsyncClient(timeout=self.timeout) as client:
27            try:
28                response = await client.request(
29                    method=method.upper(),
30                    url=url,
31                    headers=headers,
32                    content=body
33                )
34
35                return (
36                    f"Status: {response.status_code}\n"
37                    f"Headers: {dict(response.headers)}\n"
38                    f"Body: {response.text[:2000]}"
39                )
40
41            except httpx.TimeoutException:
42                return f"Error: Request timed out after {self.timeout}s"
43            except Exception as e:
44                return f"Error: {str(e)}"
45
46    def as_tool(self) -> Tool:
47        return Tool(
48            name="http_request",
49            description="Make HTTP requests to APIs and websites",
50            parameters=[
51                ToolParameter("url", "string", "The URL to request"),
52                ToolParameter("method", "string", "HTTP method", required=False),
53                ToolParameter("headers", "object", "Request headers", required=False),
54                ToolParameter("body", "string", "Request body", required=False),
55            ],
56            function=self.request,
57            category="network"
58        )

1import subprocess
2import tempfile
3import os
4from typing import Optional
5
6class CodeExecutionTool:
7    """Safely execute code."""
8
9    def __init__(
10        self,
11        timeout: int = 30,
12        allowed_languages: list[str] = None
13    ):
14        self.timeout = timeout
15        self.allowed_languages = allowed_languages or ["python"]
16
17    async def execute(
18        self,
19        code: str,
20        language: str = "python"
21    ) -> str:
22        """
23        Execute code and return output.
24
25        Args:
26            code: Code to execute
27            language: Programming language
28
29        Returns:
30            Execution output or error
31        """
32        if language not in self.allowed_languages:
33            return f"Error: Language '{language}' not allowed"
34
35        # Create temporary file
36        suffix = {
37            "python": ".py",
38            "javascript": ".js",
39            "bash": ".sh"
40        }.get(language, ".txt")
41
42        with tempfile.NamedTemporaryFile(
43            mode="w",
44            suffix=suffix,
45            delete=False
46        ) as f:
47            f.write(code)
48            temp_path = f.name
49
50        try:
51            # Get command for language
52            cmd = {
53                "python": ["python3", temp_path],
54                "javascript": ["node", temp_path],
55                "bash": ["bash", temp_path]
56            }.get(language)
57
58            if not cmd:
59                return f"Error: Unknown language '{language}'"
60
61            # Execute with timeout
62            result = subprocess.run(
63                cmd,
64                capture_output=True,
65                text=True,
66                timeout=self.timeout
67            )
68
69            output = result.stdout
70            if result.stderr:
71                output += f"\nStderr: {result.stderr}"
72
73            return output or "(No output)"
74
75        except subprocess.TimeoutExpired:
76            return f"Error: Execution timed out after {self.timeout}s"
77        except Exception as e:
78            return f"Error: {str(e)}"
79        finally:
80            os.unlink(temp_path)
81
82    def as_tool(self) -> Tool:
83        return Tool(
84            name="execute_code",
85            description="Execute code and return the output",
86            parameters=[
87                ToolParameter("code", "string", "Code to execute"),
88                ToolParameter("language", "string", "Programming language", required=False),
89            ],
90            function=self.execute,
91            category="computation",
92            requires_confirmation=True
93        )

1import os
2from pathlib import Path
3
4class FileOperationsTool:
5    """Safe file operations within allowed directories."""
6
7    def __init__(self, allowed_dirs: list[str]):
8        self.allowed_dirs = [Path(d).resolve() for d in allowed_dirs]
9
10    def _is_path_allowed(self, path: str) -> bool:
11        """Check if path is within allowed directories."""
12        resolved = Path(path).resolve()
13        return any(
14            str(resolved).startswith(str(allowed))
15            for allowed in self.allowed_dirs
16        )
17
18    async def read_file(self, path: str) -> str:
19        """Read a file's contents."""
20        if not self._is_path_allowed(path):
21            return f"Error: Path '{path}' is not in allowed directories"
22
23        try:
24            with open(path, "r") as f:
25                content = f.read()
26            return content[:10000]  # Limit size
27        except Exception as e:
28            return f"Error reading file: {str(e)}"
29
30    async def write_file(self, path: str, content: str) -> str:
31        """Write content to a file."""
32        if not self._is_path_allowed(path):
33            return f"Error: Path '{path}' is not in allowed directories"
34
35        try:
36            with open(path, "w") as f:
37                f.write(content)
38            return f"Successfully wrote {len(content)} characters to {path}"
39        except Exception as e:
40            return f"Error writing file: {str(e)}"
41
42    async def list_directory(self, path: str) -> str:
43        """List contents of a directory."""
44        if not self._is_path_allowed(path):
45            return f"Error: Path '{path}' is not in allowed directories"
46
47        try:
48            entries = []
49            for entry in os.listdir(path):
50                full_path = os.path.join(path, entry)
51                entry_type = "dir" if os.path.isdir(full_path) else "file"
52                entries.append(f"[{entry_type}] {entry}")
53            return "\n".join(entries) if entries else "(Empty directory)"
54        except Exception as e:
55            return f"Error listing directory: {str(e)}"
56
57    def get_tools(self) -> list[Tool]:
58        """Get all file operation tools."""
59        return [
60            Tool(
61                name="read_file",
62                description="Read contents of a file",
63                parameters=[ToolParameter("path", "string", "File path")],
64                function=self.read_file,
65                category="file"
66            ),
67            Tool(
68                name="write_file",
69                description="Write content to a file",
70                parameters=[
71                    ToolParameter("path", "string", "File path"),
72                    ToolParameter("content", "string", "Content to write"),
73                ],
74                function=self.write_file,
75                category="file",
76                requires_confirmation=True
77            ),
78            Tool(
79                name="list_directory",
80                description="List contents of a directory",
81                parameters=[ToolParameter("path", "string", "Directory path")],
82                function=self.list_directory,
83                category="file"
84            ),
85        ]

1class CompositeTools:
2    """Tools that combine multiple operations."""
3
4    def __init__(
5        self,
6        search_tool: WebSearchTool,
7        http_tool: HTTPTool
8    ):
9        self.search = search_tool
10        self.http = http_tool
11
12    async def research_topic(
13        self,
14        topic: str,
15        depth: str = "summary"
16    ) -> str:
17        """
18        Research a topic by searching and reading pages.
19
20        Args:
21            topic: Topic to research
22            depth: "summary" or "detailed"
23
24        Returns:
25            Research findings
26        """
27        # Step 1: Search for the topic
28        search_results = await self.search.search(topic, num_results=3)
29
30        if depth == "summary":
31            return f"Search Results:\n{search_results}"
32
33        # Step 2: Fetch top results for detailed research
34        findings = [f"Research on: {topic}\n"]
35
36        # Parse URLs from search results
37        import re
38        urls = re.findall(r'URL: (https?://\S+)', search_results)
39
40        for url in urls[:3]:
41            page_content = await self.http.request(url)
42            findings.append(f"\n--- From {url} ---\n{page_content[:1000]}")
43
44        return "\n".join(findings)
45
46    def as_tool(self) -> Tool:
47        return Tool(
48            name="research_topic",
49            description="Research a topic by searching and reading web pages",
50            parameters=[
51                ToolParameter("topic", "string", "Topic to research"),
52                ToolParameter("depth", "string", "Research depth", required=False),
53            ],
54            function=self.research_topic,
55            category="research"
56        )

1from dataclasses import dataclass
2from typing import Callable, Any
3
4@dataclass
5class PipelineStep:
6    """A step in a tool pipeline."""
7    tool: Tool
8    input_mapper: Callable[[dict], dict]  # Map previous output to input
9    output_key: str  # Key to store result
10
11class ToolPipeline:
12    """Execute a sequence of tools."""
13
14    def __init__(self, name: str, steps: list[PipelineStep]):
15        self.name = name
16        self.steps = steps
17
18    async def execute(self, initial_input: dict) -> dict:
19        """Execute all steps in sequence."""
20        context = {"input": initial_input}
21
22        for step in self.steps:
23            # Map input from context
24            tool_input = step.input_mapper(context)
25
26            # Execute tool
27            result = await step.tool.execute(**tool_input)
28
29            # Store result
30            context[step.output_key] = result
31
32        return context
33
34    def as_tool(self) -> Tool:
35        """Wrap pipeline as a single tool."""
36        async def execute_pipeline(**kwargs) -> str:
37            result = await self.execute(kwargs)
38            return str(result)
39
40        return Tool(
41            name=self.name,
42            description=f"Pipeline: {self.name}",
43            parameters=[
44                ToolParameter("input", "object", "Pipeline input")
45            ],
46            function=execute_pipeline,
47            category="pipeline"
48        )

1import asyncio
2from contextlib import asynccontextmanager
3from typing import Any
4
5class ToolSandbox:
6    """Sandbox for safe tool execution."""
7
8    def __init__(
9        self,
10        timeout: int = 30,
11        max_memory_mb: int = 512
12    ):
13        self.timeout = timeout
14        self.max_memory_mb = max_memory_mb
15
16    @asynccontextmanager
17    async def sandbox(self):
18        """Context manager for sandboxed execution."""
19        # Set resource limits (platform-specific)
20        # In production, use Docker or similar
21        try:
22            yield
23        finally:
24            # Cleanup resources
25            pass
26
27    async def execute(
28        self,
29        tool: Tool,
30        **kwargs
31    ) -> tuple[bool, str]:
32        """
33        Execute tool in sandbox.
34
35        Returns:
36            (success, result_or_error)
37        """
38        async with self.sandbox():
39            try:
40                result = await asyncio.wait_for(
41                    tool.execute(**kwargs),
42                    timeout=self.timeout
43                )
44                return True, result
45            except asyncio.TimeoutError:
46                return False, f"Tool execution timed out after {self.timeout}s"
47            except Exception as e:
48                return False, f"Tool execution error: {str(e)}"
49
50
51class ToolExecutor:
52    """Execute tools with safety measures."""
53
54    def __init__(
55        self,
56        sandbox: ToolSandbox = None,
57        confirm_callback: Callable = None
58    ):
59        self.sandbox = sandbox or ToolSandbox()
60        self.confirm_callback = confirm_callback
61
62    async def execute(
63        self,
64        tool: Tool,
65        **kwargs
66    ) -> str:
67        """Execute a tool safely."""
68
69        # Check if confirmation needed
70        if tool.requires_confirmation and self.confirm_callback:
71            confirmed = await self.confirm_callback(
72                tool.name,
73                kwargs
74            )
75            if not confirmed:
76                return "Tool execution cancelled by user"
77
78        # Execute in sandbox
79        success, result = await self.sandbox.execute(tool, **kwargs)
80
81        if not success:
82            # Log error for monitoring
83            print(f"Tool {tool.name} failed: {result}")
84
85        return result

1import time
2from collections import defaultdict
3
4class RateLimiter:
5    """Rate limit tool executions."""
6
7    def __init__(self):
8        self.limits: dict[str, tuple[int, int]] = {}  # tool -> (calls, period_seconds)
9        self.history: dict[str, list[float]] = defaultdict(list)
10
11    def set_limit(self, tool_name: str, calls: int, period_seconds: int) -> None:
12        """Set rate limit for a tool."""
13        self.limits[tool_name] = (calls, period_seconds)
14
15    def check(self, tool_name: str) -> tuple[bool, Optional[float]]:
16        """
17        Check if tool can be called.
18
19        Returns:
20            (allowed, wait_time_if_limited)
21        """
22        if tool_name not in self.limits:
23            return True, None
24
25        calls, period = self.limits[tool_name]
26        now = time.time()
27
28        # Clean old entries
29        self.history[tool_name] = [
30            t for t in self.history[tool_name]
31            if now - t < period
32        ]
33
34        if len(self.history[tool_name]) >= calls:
35            oldest = self.history[tool_name][0]
36            wait_time = period - (now - oldest)
37            return False, wait_time
38
39        return True, None
40
41    def record(self, tool_name: str) -> None:
42        """Record a tool call."""
43        self.history[tool_name].append(time.time())

1from typing import Optional
2
3class ToolRegistry:
4    """Central registry for agent tools."""
5
6    def __init__(self):
7        self.tools: dict[str, Tool] = {}
8        self.categories: dict[str, list[str]] = {}
9        self.executor = ToolExecutor()
10        self.rate_limiter = RateLimiter()
11
12    def register(self, tool: Tool) -> None:
13        """Register a tool."""
14        self.tools[tool.name] = tool
15
16        # Add to category
17        if tool.category not in self.categories:
18            self.categories[tool.category] = []
19        self.categories[tool.category].append(tool.name)
20
21    def register_many(self, tools: list[Tool]) -> None:
22        """Register multiple tools."""
23        for tool in tools:
24            self.register(tool)
25
26    def get(self, name: str) -> Optional[Tool]:
27        """Get a tool by name."""
28        return self.tools.get(name)
29
30    def get_by_category(self, category: str) -> list[Tool]:
31        """Get all tools in a category."""
32        names = self.categories.get(category, [])
33        return [self.tools[n] for n in names]
34
35    def get_all_schemas(self) -> list[dict]:
36        """Get API schemas for all tools."""
37        return [tool.to_api_schema() for tool in self.tools.values()]
38
39    def get_schemas_for_task(self, task: str) -> list[dict]:
40        """Get relevant tools for a task (could use LLM selection)."""
41        # Simple implementation: return all tools
42        # Advanced: use embeddings or LLM to select relevant tools
43        return self.get_all_schemas()
44
45    async def execute(self, tool_name: str, **kwargs) -> str:
46        """Execute a tool by name."""
47        tool = self.get(tool_name)
48        if not tool:
49            return f"Error: Unknown tool '{tool_name}'"
50
51        # Check rate limit
52        allowed, wait_time = self.rate_limiter.check(tool_name)
53        if not allowed:
54            return f"Rate limited. Try again in {wait_time:.1f}s"
55
56        # Execute
57        result = await self.executor.execute(tool, **kwargs)
58
59        # Record for rate limiting
60        self.rate_limiter.record(tool_name)
61
62        return result
63
64    def describe(self) -> str:
65        """Get human-readable description of all tools."""
66        lines = ["Available Tools:"]
67        for category, tool_names in self.categories.items():
68            lines.append(f"\n## {category.title()}")
69            for name in tool_names:
70                tool = self.tools[name]
71                lines.append(f"  - {tool.name}: {tool.description}")
72        return "\n".join(lines)

1# Create and populate registry
2registry = ToolRegistry()
3
4# Register tools
5registry.register(WebSearchTool(api_key="...").as_tool())
6registry.register(HTTPTool().as_tool())
7registry.register(CodeExecutionTool().as_tool())
8registry.register_many(FileOperationsTool(["./workspace"]).get_tools())
9
10# Set rate limits
11registry.rate_limiter.set_limit("web_search", calls=10, period_seconds=60)
12registry.rate_limiter.set_limit("execute_code", calls=5, period_seconds=60)
13
14# Use in agent
15class AgentWithRegistry:
16    def __init__(self, registry: ToolRegistry):
17        self.registry = registry
18
19    def get_tools(self) -> list[dict]:
20        return self.registry.get_all_schemas()
21
22    async def execute_tool(self, name: str, input: dict) -> str:
23        return await self.registry.execute(name, **input)