"""Tool execution engine for agent tool calls.""" import ast import operator import random from datetime import datetime from typing import Any, Dict, List, Union from zoneinfo import ZoneInfo class SafeCalculator: """Safe calculator using AST parsing instead of eval().""" # Allowed operations operators = { ast.Add: operator.add, ast.Sub: operator.sub, ast.Mult: operator.mul, ast.Div: operator.truediv, ast.Pow: operator.pow, ast.USub: operator.neg, } def evaluate(self, expression: str) -> float: """Safely evaluate a mathematical expression. Args: expression: Mathematical expression string Returns: Result of the calculation Raises: ValueError: If expression is invalid or uses disallowed operations """ try: tree = ast.parse(expression, mode='eval') return self._eval_node(tree.body) except Exception as e: raise ValueError(f"Invalid expression: {str(e)}") def _eval_node(self, node) -> Union[float, int]: """Recursively evaluate AST nodes. Args: node: AST node to evaluate Returns: Numeric result """ if isinstance(node, ast.Constant): return node.value elif isinstance(node, ast.Num): # For Python < 3.8 compatibility return node.n elif isinstance(node, ast.BinOp): left = self._eval_node(node.left) right = self._eval_node(node.right) op_type = type(node.op) if op_type not in self.operators: raise ValueError(f"Unsupported operation: {op_type.__name__}") return self.operators[op_type](left, right) elif isinstance(node, ast.UnaryOp): operand = self._eval_node(node.operand) op_type = type(node.op) if op_type not in self.operators: raise ValueError(f"Unsupported operation: {op_type.__name__}") return self.operators[op_type](operand) else: raise ValueError(f"Unsupported node type: {type(node).__name__}") class ToolExecutor: """Executes tool calls from agent models.""" def __init__(self): """Initialize tool executor.""" self.execution_history: List[Dict[str, Any]] = [] self.calculator = SafeCalculator() def execute(self, tool_name: str, arguments: Dict[str, Any]) -> Dict[str, Any]: """Execute a tool call. Args: tool_name: Name of the tool to execute arguments: Tool arguments Returns: Dictionary with 'success', 'result', and optional 'error' keys """ try: if tool_name == "calculator": result = self._execute_calculator(arguments) elif tool_name == "get_weather": result = self._execute_get_weather(arguments) elif tool_name == "web_search": result = self._execute_web_search(arguments) elif tool_name == "get_current_time": result = self._execute_get_current_time(arguments) elif tool_name == "convert_units": result = self._execute_convert_units(arguments) else: return { "success": False, "error": f"Unknown tool: {tool_name}", } # Log execution self.execution_history.append( { "tool": tool_name, "arguments": arguments, "result": result, "timestamp": datetime.now().isoformat(), } ) return {"success": True, "result": result} except Exception as e: error_msg = f"Error executing {tool_name}: {str(e)}" return {"success": False, "error": error_msg} def _execute_calculator(self, args: Dict[str, Any]) -> str: """Execute calculator tool. Args: args: Arguments containing 'expression' Returns: Calculation result as string """ expression = args.get("expression", "") if not expression: raise ValueError("No expression provided") result = self.calculator.evaluate(expression) return f"Result: {result}" def _execute_get_weather(self, args: Dict[str, Any]) -> str: """Execute weather lookup tool (simulated). Args: args: Arguments containing 'city' and optional 'unit' Returns: Weather information as string """ city = args.get("city", "") unit = args.get("unit", "celsius") if not city: raise ValueError("No city provided") # Simulated weather data temp_c = random.randint(5, 30) temp_f = int(temp_c * 9 / 5 + 32) temp = temp_f if unit == "fahrenheit" else temp_c unit_symbol = "°F" if unit == "fahrenheit" else "°C" conditions = random.choice(["Sunny", "Partly Cloudy", "Cloudy", "Rainy", "Clear"]) humidity = random.randint(40, 90) return ( f"Weather in {city}:\n" f"Temperature: {temp}{unit_symbol}\n" f"Conditions: {conditions}\n" f"Humidity: {humidity}%" ) def _execute_web_search(self, args: Dict[str, Any]) -> str: """Execute web search tool (simulated). Args: args: Arguments containing 'query' and optional 'max_results' Returns: Search results as string """ query = args.get("query", "") max_results = args.get("max_results", 3) if not query: raise ValueError("No query provided") # Simulated search results results = [] for i in range(min(max_results, 3)): results.append( f"{i+1}. Result for '{query}': This is a simulated search result " f"containing relevant information about {query}." ) return "Search results:\n" + "\n".join(results) def _execute_get_current_time(self, args: Dict[str, Any]) -> str: """Execute current time tool. Args: args: Arguments containing optional 'timezone' and 'format' Returns: Current time as string """ timezone_str = args.get("timezone", "UTC") time_format = args.get("format", "24h") try: tz = ZoneInfo(timezone_str) now = datetime.now(tz) if time_format == "12h": time_str = now.strftime("%I:%M:%S %p") elif time_format == "iso": time_str = now.isoformat() else: # 24h time_str = now.strftime("%H:%M:%S") date_str = now.strftime("%Y-%m-%d") return f"Current time in {timezone_str}:\nDate: {date_str}\nTime: {time_str}" except Exception as e: raise ValueError(f"Invalid timezone: {str(e)}") def _execute_convert_units(self, args: Dict[str, Any]) -> str: """Execute unit conversion tool. Args: args: Arguments containing 'value', 'from_unit', and 'to_unit' Returns: Conversion result as string """ value = args.get("value") from_unit = args.get("from_unit", "").lower() to_unit = args.get("to_unit", "").lower() if value is None: raise ValueError("No value provided") if not from_unit or not to_unit: raise ValueError("Both from_unit and to_unit required") # Conversion factors (base unit: meters, kilograms, celsius) conversions = { # Distance ("km", "miles"): 0.621371, ("miles", "km"): 1.60934, ("m", "feet"): 3.28084, ("feet", "m"): 0.3048, ("cm", "inches"): 0.393701, ("inches", "cm"): 2.54, # Weight ("kg", "lbs"): 2.20462, ("lbs", "kg"): 0.453592, ("g", "oz"): 0.035274, ("oz", "g"): 28.3495, # Temperature ("celsius", "fahrenheit"): lambda x: x * 9 / 5 + 32, ("fahrenheit", "celsius"): lambda x: (x - 32) * 5 / 9, } # Check for same unit if from_unit == to_unit: return f"{value} {from_unit} = {value} {to_unit}" # Find conversion key = (from_unit, to_unit) if key in conversions: factor = conversions[key] if callable(factor): result = factor(value) else: result = value * factor return f"{value} {from_unit} = {result:.2f} {to_unit}" else: raise ValueError( f"Conversion from {from_unit} to {to_unit} not supported. " f"Available conversions: {list(conversions.keys())}" ) def get_execution_history(self) -> List[Dict[str, Any]]: """Get tool execution history. Returns: List of execution records """ return self.execution_history def clear_history(self): """Clear execution history.""" self.execution_history = []