tools/tool_engine.py

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"})