""" tools/whitelist_tool.py — Web domain whitelist management for agents (async). """ from __future__ import annotations from ..database import web_whitelist_store from .base import BaseTool, ToolResult class WhitelistTool(BaseTool): name = "whitelist" description = ( "Manage the Tier-1 web domain whitelist. " "Use this when the user asks to add, remove, or list whitelisted websites. " "Operations: list_domains, add_domain, remove_domain." ) input_schema = { "type": "object", "properties": { "operation": { "type": "string", "enum": ["list_domains", "add_domain", "remove_domain"], "description": "Operation to perform", }, "domain": { "type": "string", "description": "Domain to add or remove (e.g. 'nrk.no'). Required for add_domain and remove_domain.", }, "note": { "type": "string", "description": "Optional description of why this domain is whitelisted.", }, }, "required": ["operation"], } requires_confirmation = False allowed_in_scheduled_tasks = True async def execute(self, operation: str, domain: str = "", note: str = "", **kwargs) -> ToolResult: if operation == "list_domains": domains = await web_whitelist_store.list() return ToolResult(success=True, data={"domains": domains, "count": len(domains)}) if operation == "add_domain": if not domain: return ToolResult(success=False, error="domain is required for add_domain") await web_whitelist_store.add(domain.strip(), note) return ToolResult(success=True, data={"added": domain.strip(), "note": note}) if operation == "remove_domain": if not domain: return ToolResult(success=False, error="domain is required for remove_domain") removed = await web_whitelist_store.remove(domain.strip()) if not removed: return ToolResult(success=False, error=f"Domain '{domain}' not found in whitelist") return ToolResult(success=True, data={"removed": domain.strip()}) return ToolResult(success=False, error=f"Unknown operation: {operation}")