"""Tool access control: central policy for which agent may call which tools. In ADVISORY mode, denials are logged as advisories and the action proceeds. The system learns from outcomes rather than being caged. """ from fusionagi._logger import logger from fusionagi.schemas.audit import GovernanceMode class AccessControl: """Policy: (agent_id, tool_name, task_id) -> allowed. In ADVISORY mode (default), denied access is logged but permitted. """ def __init__(self, mode: GovernanceMode = GovernanceMode.ADVISORY) -> None: self._deny: set[tuple[str, str]] = set() self._task_tools: dict[str, set[str]] = {} self._mode = mode def deny(self, agent_id: str, tool_name: str) -> None: """Register a denial rule for agent/tool pair.""" self._deny.add((agent_id, tool_name)) def allow_tools_for_task(self, task_id: str, tool_names: list[str]) -> None: """Set allowed tools for a task (empty = all allowed).""" self._task_tools[task_id] = set(tool_names) def allowed(self, agent_id: str, tool_name: str, task_id: str | None = None) -> bool: """Return True if agent may call tool. In ADVISORY mode, always returns True but logs advisory if a rule would have denied the action. """ if (agent_id, tool_name) in self._deny: if self._mode == GovernanceMode.ADVISORY: logger.info( "AccessControl advisory: agent/tool denied (proceeding)", extra={"agent_id": agent_id, "tool_name": tool_name, "mode": "advisory"}, ) return True return False if task_id and task_id in self._task_tools: if tool_name not in self._task_tools[task_id]: if self._mode == GovernanceMode.ADVISORY: logger.info( "AccessControl advisory: tool not in task allowlist (proceeding)", extra={"agent_id": agent_id, "tool_name": tool_name, "task_id": task_id, "mode": "advisory"}, ) return True return False return True