from typing import Dict from schemas.enums import SafetyLevel from schemas.models import ToolDefinition, ToolRequest, ToolResponse class ToolPermissionEngine: """Manages catalog of colony tools and validates execution permissions against safety guardrails.""" def __init__(self): self.catalog: Dict[str, ToolDefinition] = { "Playwright": ToolDefinition( name="Playwright", category="Browser", description="Automated web browser navigation and interaction", permission_level=SafetyLevel.INTERACTIVE_FORM ), "HTTP": ToolDefinition( name="HTTP", category="Network", description="Send HTTP/REST requests to external endpoints", permission_level=SafetyLevel.READ_ONLY ), "Vision": ToolDefinition( name="Vision", category="Perception", description="Visual OCR and screen element analysis", permission_level=SafetyLevel.READ_ONLY ), "SQLite": ToolDefinition( name="SQLite", category="Database", description="Read and store local structured operational data", permission_level=SafetyLevel.READ_ONLY ), "Memory": ToolDefinition( name="Memory", category="Memory", description="Search and persist vector/text memories", permission_level=SafetyLevel.READ_ONLY ), "Embeddings": ToolDefinition( name="Embeddings", category="AI", description="Generate semantic vector embeddings", permission_level=SafetyLevel.READ_ONLY ), "Filesystem": ToolDefinition( name="Filesystem", category="System", description="Read and write safe workspace files", permission_level=SafetyLevel.SENSITIVE_ACTION ), "Search": ToolDefinition( name="Search", category="Research", description="Execute web search and citation retrieval", permission_level=SafetyLevel.READ_ONLY ), } def evaluate_request(self, req: ToolRequest) -> ToolResponse: tool = self.catalog.get(req.tool_name) if not tool: return ToolResponse(tool_name=req.tool_name, allowed=False, reason=f"Tool '{req.tool_name}' not registered in catalog.") if not tool.enabled: return ToolResponse(tool_name=req.tool_name, allowed=False, reason=f"Tool '{req.tool_name}' is currently disabled.") if tool.permission_level == SafetyLevel.DESTRUCTIVE_BLOCKED: return ToolResponse(tool_name=req.tool_name, allowed=False, reason=f"Action blocked by safety policy: Destructive operations prohibited.") return ToolResponse(tool_name=req.tool_name, allowed=True, reason="Permission granted.", result={"status": "READY"})