Spaces:
Running on Zero
Running on Zero
| """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}" | |