378d599125
Multi-agent AI automation system with shared message bus, specialized roles (coder/researcher/reviewer), and deny-by-default security. - Config system with Pydantic validation and YAML loading - Async message bus with inter-agent delegation - LLM providers: Anthropic (Claude) and LiteLLM (DeepSeek/Kimi/MiniMax) - Tool system: registry, builtins (file/bash/web), approval engine, MCP client - Agent engine with tool-calling loop and orchestrator for multi-agent management - CLI channel (REPL) and Discord channel - Docker + Dockge deployment config - Typer CLI: chat, serve, status, agents commands Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
70 lines
2.4 KiB
Python
70 lines
2.4 KiB
Python
"""Tool approval engine — deny-by-default, per-agent policies."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from enum import Enum
|
|
from typing import Any
|
|
|
|
from loguru import logger
|
|
|
|
|
|
class ApprovalPolicy(Enum):
|
|
AUTO_APPROVE = "auto_approve"
|
|
REQUIRE_APPROVAL = "require_approval"
|
|
DENY = "deny"
|
|
|
|
|
|
class ApprovalEngine:
|
|
"""Deny-by-default tool approval."""
|
|
|
|
def __init__(
|
|
self,
|
|
auto_approve: list[str] | None = None,
|
|
require_approval: list[str] | None = None,
|
|
interactive: bool = True,
|
|
) -> None:
|
|
self._auto_approve = set(auto_approve or [])
|
|
self._require_approval = set(require_approval or [])
|
|
self._interactive = interactive
|
|
|
|
def get_policy(self, tool_name: str) -> ApprovalPolicy:
|
|
"""Get the approval policy for a tool."""
|
|
# MCP tools inherit from the mcp_* prefix pattern
|
|
base_name = tool_name.split("_", 1)[0] if tool_name.startswith("mcp_") else tool_name
|
|
|
|
if tool_name in self._auto_approve or base_name in self._auto_approve:
|
|
return ApprovalPolicy.AUTO_APPROVE
|
|
if tool_name in self._require_approval or base_name in self._require_approval:
|
|
return ApprovalPolicy.REQUIRE_APPROVAL
|
|
return ApprovalPolicy.DENY
|
|
|
|
async def check(self, tool_name: str, arguments: dict[str, Any]) -> bool:
|
|
"""Check if a tool call is approved. Returns True if approved."""
|
|
policy = self.get_policy(tool_name)
|
|
|
|
if policy == ApprovalPolicy.AUTO_APPROVE:
|
|
return True
|
|
|
|
if policy == ApprovalPolicy.DENY:
|
|
logger.warning(f"Tool '{tool_name}' denied by policy")
|
|
return False
|
|
|
|
# REQUIRE_APPROVAL
|
|
if not self._interactive:
|
|
logger.warning(f"Tool '{tool_name}' requires approval but running non-interactively — denied")
|
|
return False
|
|
|
|
# In interactive mode, prompt the user
|
|
logger.info(f"Tool '{tool_name}' requires approval. Args: {arguments}")
|
|
return await self._prompt_user(tool_name, arguments)
|
|
|
|
async def _prompt_user(self, tool_name: str, arguments: dict[str, Any]) -> bool:
|
|
"""Prompt user for tool approval (interactive mode)."""
|
|
print(f"\n[APPROVAL REQUIRED] Tool: {tool_name}")
|
|
print(f" Arguments: {arguments}")
|
|
try:
|
|
answer = input(" Allow? [y/N]: ").strip().lower()
|
|
return answer in ("y", "yes")
|
|
except (EOFError, KeyboardInterrupt):
|
|
return False
|