""" Risk classification for tool calls. Categorizes tools by risk level and determines if confirmation is required. """ from enum import Enum from typing import Dict, Set, Optional class RiskLevel(Enum): """Risk levels for tool calls.""" LOW = "low" # No confirmation needed MEDIUM = "medium" # Optional confirmation HIGH = "high" # Confirmation required CRITICAL = "critical" # Explicit confirmation required class RiskClassifier: """Classifies tool calls by risk level.""" def __init__(self): """Initialize risk classifier with tool risk mappings.""" # High-risk tools that require confirmation self.high_risk_tools: Set[str] = { "send_email", "create_calendar_event", "update_calendar_event", "delete_calendar_event", "set_smart_home_scene", "toggle_smart_device", "adjust_thermostat", "delete_note", "delete_task", } # Medium-risk tools (optional confirmation) self.medium_risk_tools: Set[str] = { "update_task_status", # Moving tasks between columns "append_to_note", # Modifying existing notes "create_reminder", # Creating reminders } # Low-risk tools (no confirmation needed) self.low_risk_tools: Set[str] = { "get_current_time", "get_date", "get_timezone_info", "convert_timezone", "get_weather", "list_timers", "list_tasks", "list_notes", "read_note", "search_notes", "create_timer", # Timers are low-risk "create_note", # Creating new notes is low-risk "add_task", # Adding tasks is low-risk } # Critical-risk tools (explicit confirmation required) self.critical_risk_tools: Set[str] = { "send_email", # Sending emails is critical "delete_calendar_event", # Deleting calendar events "set_smart_home_scene", # Smart home control } def classify_risk(self, tool_name: str, **kwargs) -> RiskLevel: """ Classify risk level for a tool call. Args: tool_name: Name of the tool **kwargs: Tool-specific parameters that might affect risk Returns: RiskLevel enum """ # Check critical first if tool_name in self.critical_risk_tools: return RiskLevel.CRITICAL # Check high risk if tool_name in self.high_risk_tools: return RiskLevel.HIGH # Check medium risk if tool_name in self.medium_risk_tools: # Can be elevated to HIGH based on context if self._is_high_risk_context(tool_name, **kwargs): return RiskLevel.HIGH return RiskLevel.MEDIUM # Default to low risk if tool_name in self.low_risk_tools: return RiskLevel.LOW # Unknown tool - default to medium risk (safe default) return RiskLevel.MEDIUM def _is_high_risk_context(self, tool_name: str, **kwargs) -> bool: """Check if context elevates risk level.""" # Example: Updating task to "done" might be higher risk if tool_name == "update_task_status": new_status = kwargs.get("status", "").lower() if new_status in ["done", "cancelled"]: return True # Example: Appending to important notes if tool_name == "append_to_note": note_path = kwargs.get("note_path", "") if "important" in note_path.lower() or "critical" in note_path.lower(): return True return False def requires_confirmation(self, tool_name: str, **kwargs) -> bool: """ Check if tool call requires confirmation. Args: tool_name: Name of the tool **kwargs: Tool-specific parameters Returns: True if confirmation is required """ risk = self.classify_risk(tool_name, **kwargs) return risk in [RiskLevel.HIGH, RiskLevel.CRITICAL] def get_confirmation_message(self, tool_name: str, **kwargs) -> str: """ Generate confirmation message for tool call. Args: tool_name: Name of the tool **kwargs: Tool-specific parameters Returns: Confirmation message """ risk = self.classify_risk(tool_name, **kwargs) if risk == RiskLevel.CRITICAL: return f"⚠️ CRITICAL ACTION: I'm about to {self._describe_action(tool_name, **kwargs)}. This cannot be undone. Do you want to proceed? (Yes/No)" elif risk == RiskLevel.HIGH: return f"⚠️ I'm about to {self._describe_action(tool_name, **kwargs)}. Do you want to proceed? (Yes/No)" else: return f"I'm about to {self._describe_action(tool_name, **kwargs)}. Proceed? (Yes/No)" def _describe_action(self, tool_name: str, **kwargs) -> str: """Generate human-readable description of action.""" descriptions = { "send_email": f"send an email to {kwargs.get('to', 'recipient')}", "create_calendar_event": f"create a calendar event: {kwargs.get('title', 'event')}", "update_calendar_event": f"update calendar event: {kwargs.get('event_id', 'event')}", "delete_calendar_event": f"delete calendar event: {kwargs.get('event_id', 'event')}", "set_smart_home_scene": f"set smart home scene: {kwargs.get('scene_name', 'scene')}", "toggle_smart_device": f"toggle device: {kwargs.get('device_name', 'device')}", "adjust_thermostat": f"adjust thermostat to {kwargs.get('temperature', 'temperature')}°", "delete_note": f"delete note: {kwargs.get('note_path', 'note')}", "delete_task": f"delete task: {kwargs.get('task_path', 'task')}", "update_task_status": f"update task status to {kwargs.get('status', 'status')}", "append_to_note": f"append to note: {kwargs.get('note_path', 'note')}", "create_reminder": f"create a reminder: {kwargs.get('message', 'reminder')}", } return descriptions.get(tool_name, f"execute {tool_name}") # Global classifier instance _classifier = RiskClassifier() def get_classifier() -> RiskClassifier: """Get the global risk classifier instance.""" return _classifier