"""Tool executor — manages and executes agent tools.""" import json import math import time import asyncio from datetime import datetime from typing import Any, Callable, Optional import logging logger = logging.getLogger("synapse.tools") class Tool: """Represents a callable tool for the agent.""" def __init__(self, name: str, description: str, parameters: dict, handler: Callable, requires_gpu: bool = False): self.name = name self.description = description self.parameters = parameters self.handler = handler self.requires_gpu = requires_gpu class ToolExecutor: """Registry and executor for all agent tools.""" def __init__(self): self._tools: dict[str, Tool] = {} self._register_builtins() def _register_builtins(self): self.register(Tool( name="web_search", description="Search the internet for current information", parameters={"query": {"type": "string", "description": "Search query"}}, handler=self._web_search, )) self.register(Tool( name="calculator", description="Perform mathematical calculations", parameters={"expression": {"type": "string", "description": "Math expression to evaluate"}}, handler=self._calculator, )) self.register(Tool( name="get_time", description="Get current date and time", parameters={}, handler=self._get_time, )) self.register(Tool( name="get_weather", description="Get weather information for a location", parameters={"location": {"type": "string", "description": "City name"}}, handler=self._get_weather, )) self.register(Tool( name="summarizer", description="Summarize text content", parameters={"text": {"type": "string", "description": "Text to summarize"}}, handler=self._summarize, )) self.register(Tool( name="translate", description="Translate text between languages", parameters={ "text": {"type": "string", "description": "Text to translate"}, "target_lang": {"type": "string", "description": "Target language code"}, }, handler=self._translate, )) self.register(Tool( name="unit_converter", description="Convert between units of measurement", parameters={ "value": {"type": "number", "description": "Value to convert"}, "from_unit": {"type": "string", "description": "Source unit"}, "to_unit": {"type": "string", "description": "Target unit"}, }, handler=self._unit_convert, )) self.register(Tool( name="python_exec", description="Execute Python code and return the result", parameters={"code": {"type": "string", "description": "Python code to execute"}}, handler=self._python_exec, )) self.register(Tool( name="read_file", description="Read contents of a file", parameters={"path": {"type": "string", "description": "File path"}}, handler=self._read_file, )) def register(self, tool: Tool): self._tools[tool.name] = tool def get_tool_definitions(self) -> list[dict]: return [ { "name": t.name, "description": t.description, "parameters": t.parameters, } for t in self._tools.values() ] async def execute(self, tool_name: str, **kwargs) -> dict: if tool_name not in self._tools: return {"error": f"Unknown tool: {tool_name}"} tool = self._tools[tool_name] logger.info(f"Executing tool: {tool_name} with args: {list(kwargs.keys())}") try: start = time.time() if asyncio.iscoroutinefunction(tool.handler): result = await tool.handler(**kwargs) else: result = tool.handler(**kwargs) elapsed = time.time() - start return { "success": True, "result": result, "tool": tool_name, "elapsed": round(elapsed, 3), } except Exception as e: logger.error(f"Tool {tool_name} failed: {e}") return {"success": False, "error": str(e), "tool": tool_name} async def _web_search(self, query: str = "") -> str: try: from duckduckgo_search import DDGS with DDGS() as ddgs: results = list(ddgs.text(query, max_results=5)) if not results: return "No search results found." parts = [] for r in results: parts.append(f"**{r.get('title', '')}**\n{r.get('body', '')}\n{r.get('href', '')}") return "\n\n".join(parts) except Exception as e: return f"Search error: {e}" def _calculator(self, expression: str = "") -> str: allowed = set("0123456789+-*/.() %") sanitized = "".join(c for c in expression if c in allowed) if not sanitized: return "Invalid expression" try: result = eval(sanitized, {"__builtins__": {}}, {"math": math}) return str(result) except Exception as e: return f"Calculation error: {e}" def _get_time(self) -> str: now = datetime.now() return now.strftime("%Y-%m-%d %H:%M:%S %A") async def _get_weather(self, location: str = "") -> str: return f"Weather data for {location} — check a weather API for live data." def _summarize(self, text: str = "") -> str: sentences = text.split(". ") if len(sentences) <= 3: return text summary = ". ".join(sentences[:3]) + "." return summary def _translate(self, text: str = "", target_lang: str = "en") -> str: return f"[Translation to {target_lang}] {text}" def _unit_convert(self, value: float = 0, from_unit: str = "", to_unit: str = "") -> str: conversions = { ("km", "mi"): 0.621371, ("mi", "km"): 1.60934, ("kg", "lb"): 2.20462, ("lb", "kg"): 0.453592, ("c", "f"): lambda x: x * 9/5 + 32, ("f", "c"): lambda x: (x - 32) * 5/9, } key = (from_unit.lower(), to_unit.lower()) if key in conversions: conv = conversions[key] if callable(conv): result = conv(value) else: result = value * conv return f"{value} {from_unit} = {result:.4f} {to_unit}" return f"Conversion {from_unit} -> {to_unit} not supported" def _python_exec(self, code: str = "") -> str: try: import io import contextlib buffer = io.StringIO() with contextlib.redirect_stdout(buffer): exec(code, {"__builtins__": __builtins__}, {}) output = buffer.getvalue() return output if output else "Code executed successfully (no output)" except Exception as e: return f"Execution error: {e}" def _read_file(self, path: str = "") -> str: try: from pathlib import Path p = Path(path) if not p.exists(): return f"File not found: {path}" if p.stat().st_size > 1_000_000: return "File too large to read" return p.read_text(errors="replace") except Exception as e: return f"Error reading file: {e}"