import ast import operator from collections.abc import Callable from dataclasses import dataclass @dataclass class Tool: name: str description: str parameters: dict fn: Callable[..., str] class ToolRegistry: def __init__(self) -> None: self._tools: dict[str, Tool] = {} def register(self, tool: Tool) -> None: self._tools[tool.name] = tool def to_openai_schema(self) -> list[dict]: return [ { "type": "function", "function": { "name": t.name, "description": t.description, "parameters": t.parameters, }, } for t in self._tools.values() ] def execute(self, name: str, args: dict) -> str: if name not in self._tools: return f"Error: unknown tool '{name}'" try: return self._tools[name].fn(**args) except Exception as e: # tool failures must not crash the agent loop return f"Error executing {name}: {e}" _SAFE_OPS = { ast.Add: operator.add, ast.Sub: operator.sub, ast.Mult: operator.mul, ast.Div: operator.truediv, ast.Pow: operator.pow, ast.Mod: operator.mod, ast.USub: operator.neg, } def _safe_eval(node: ast.AST) -> int | float: if isinstance(node, ast.Constant) and isinstance(node.value, (int, float)): return node.value if isinstance(node, ast.BinOp) and type(node.op) in _SAFE_OPS: left = _safe_eval(node.left) right = _safe_eval(node.right) if isinstance(node.op, ast.Pow) and abs(right) > 100: raise ValueError("exponent too large") return _SAFE_OPS[type(node.op)](left, right) if isinstance(node, ast.UnaryOp) and type(node.op) in _SAFE_OPS: return _SAFE_OPS[type(node.op)](_safe_eval(node.operand)) raise ValueError("unsupported expression") def calculate(expression: str) -> str: try: tree = ast.parse(expression, mode="eval") return str(_safe_eval(tree.body)) except Exception as e: return f"Error: {e}" def make_calculator() -> Tool: return Tool( name="calculator", description="Evaluate a basic arithmetic expression, e.g. '2 * (3 + 4)'.", parameters={ "type": "object", "properties": {"expression": {"type": "string"}}, "required": ["expression"], }, fn=lambda expression: calculate(expression), ) def web_search(query: str, max_results: int = 3) -> str: try: from ddgs import DDGS except ImportError: # package was renamed; support both from duckduckgo_search import DDGS # type: ignore results = [] with DDGS() as ddgs: for r in ddgs.text(query, max_results=max_results): results.append(f"- {r.get('title', '')}: {r.get('body', '')}") return "\n".join(results) if results else "No results found." def make_web_search() -> Tool: return Tool( name="web_search", description="Search the web for current information. Returns the top results.", parameters={ "type": "object", "properties": {"query": {"type": "string"}}, "required": ["query"], }, fn=lambda query: web_search(query), )