744c149300
Some checks failed
Build and Deploy Docs / build-and-deploy (push) Has been cancelled
- Created logs/ structure (sessions, operations, incidents) - Added session-start/log/end scripts - Installed Git hooks for auto-logging commits/pushes - Added shell integration for zsh - Created CHANGELOG.md - Documented today's session (2026-01-10)
92 lines
2.4 KiB
Python
92 lines
2.4 KiB
Python
import yaml
|
|
import os
|
|
from models import ToolDefinition
|
|
from typing import Dict, Optional
|
|
|
|
class ToolRegistry:
|
|
"""Tool registry for Phase 3 (config-based)"""
|
|
|
|
def __init__(self, config_path: str = "config.yaml"):
|
|
self.tools: Dict[str, ToolDefinition] = {}
|
|
self._load_config(config_path)
|
|
|
|
def _load_config(self, config_path: str):
|
|
"""Load tool definitions from YAML"""
|
|
if not os.path.exists(config_path):
|
|
print(f"⚠️ Config file not found: {config_path}")
|
|
return
|
|
|
|
with open(config_path, 'r') as f:
|
|
config = yaml.safe_load(f)
|
|
|
|
for tool_config in config.get('tools', []):
|
|
tool_def = ToolDefinition(**tool_config)
|
|
self.tools[tool_def.id] = tool_def
|
|
print(f"✅ Loaded tool: {tool_def.id} ({tool_def.executor})")
|
|
|
|
print(f"📋 Total tools loaded: {len(self.tools)}")
|
|
|
|
def get_tool(self, tool_id: str) -> Optional[ToolDefinition]:
|
|
"""Get tool definition by ID"""
|
|
return self.tools.get(tool_id)
|
|
|
|
def list_tools(self, agent_id: Optional[str] = None) -> list[ToolDefinition]:
|
|
"""
|
|
List available tools
|
|
|
|
If agent_id provided, filter by allowed_agents
|
|
"""
|
|
tools = list(self.tools.values())
|
|
|
|
if agent_id:
|
|
tools = [
|
|
tool for tool in tools
|
|
if tool.enabled and (
|
|
tool.allowed_agents is None or
|
|
agent_id in tool.allowed_agents
|
|
)
|
|
]
|
|
else:
|
|
tools = [tool for tool in tools if tool.enabled]
|
|
|
|
return tools
|
|
|
|
def check_permission(self, tool_id: str, agent_id: str) -> tuple[bool, Optional[str]]:
|
|
"""
|
|
Check if agent has permission to use tool
|
|
|
|
Returns: (allowed: bool, reason: str | None)
|
|
"""
|
|
tool = self.get_tool(tool_id)
|
|
|
|
if not tool:
|
|
return False, f"Tool not found: {tool_id}"
|
|
|
|
if not tool.enabled:
|
|
return False, f"Tool is disabled: {tool_id}"
|
|
|
|
if tool.allowed_agents is None:
|
|
# No restrictions
|
|
return True, None
|
|
|
|
if agent_id not in tool.allowed_agents:
|
|
return False, f"Agent {agent_id} not in allowlist for {tool_id}"
|
|
|
|
return True, None
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|