tools/tool_executor.py

"""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 = []