31 lines
1.2 KiB
Python
31 lines
1.2 KiB
Python
"""Tool access control: central policy for which agent may call which tools.
|
|
|
|
Optional; not wired to Executor or Orchestrator by default. Wire by passing
|
|
an AccessControl instance and checking allowed(agent_id, tool_name, task_id)
|
|
before tool invocation.
|
|
"""
|
|
|
|
|
|
class AccessControl:
|
|
"""Policy: (agent_id, tool_name, task_id) -> allowed."""
|
|
|
|
def __init__(self) -> None:
|
|
self._deny: set[tuple[str, str]] = set()
|
|
self._task_tools: dict[str, set[str]] = {}
|
|
|
|
def deny(self, agent_id: str, tool_name: str) -> None:
|
|
"""Deny agent from using tool (global)."""
|
|
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 (optionally for this task)."""
|
|
if (agent_id, tool_name) in self._deny:
|
|
return False
|
|
if task_id and task_id in self._task_tools:
|
|
return tool_name in self._task_tools[task_id]
|
|
return True
|