70 lines
2.5 KiB
Python
70 lines
2.5 KiB
Python
"""Tool registry: register tools by name; resolve by name and check permissions."""
|
|
|
|
from typing import Any, Callable
|
|
|
|
# Tool schema: name, description, parameters (JSON Schema), permission_scope
|
|
# Invocation: (args: dict) -> result
|
|
ToolFn = Callable[..., Any]
|
|
|
|
|
|
class ToolDef:
|
|
"""Tool definition: name, description, parameters schema, permission scope, timeout."""
|
|
|
|
def __init__(
|
|
self,
|
|
name: str,
|
|
description: str,
|
|
fn: ToolFn,
|
|
parameters_schema: dict[str, Any] | None = None,
|
|
permission_scope: str | list[str] = "*",
|
|
timeout_seconds: float = 30.0,
|
|
manufacturing: bool = False,
|
|
) -> None:
|
|
self.name = name
|
|
self.description = description
|
|
self.fn = fn
|
|
self.parameters_schema = parameters_schema or {"type": "object", "properties": {}}
|
|
self.permission_scope = permission_scope if isinstance(permission_scope, list) else [permission_scope]
|
|
self.timeout_seconds = timeout_seconds
|
|
self.manufacturing = manufacturing
|
|
|
|
def to_schema(self) -> dict[str, Any]:
|
|
"""JSON Schema for this tool."""
|
|
return {
|
|
"name": self.name,
|
|
"description": self.description,
|
|
"parameters": self.parameters_schema,
|
|
"permission_scope": self.permission_scope,
|
|
"timeout_seconds": self.timeout_seconds,
|
|
"manufacturing": self.manufacturing,
|
|
}
|
|
|
|
|
|
class ToolRegistry:
|
|
"""Register and resolve tools by name; check agent permissions."""
|
|
|
|
def __init__(self) -> None:
|
|
self._tools: dict[str, ToolDef] = {}
|
|
|
|
def register(self, tool: ToolDef) -> None:
|
|
"""Register a tool by name."""
|
|
self._tools[tool.name] = tool
|
|
|
|
def get(self, name: str) -> ToolDef | None:
|
|
"""Return tool definition by name or None."""
|
|
return self._tools.get(name)
|
|
|
|
def list_tools(self) -> list[dict[str, Any]]:
|
|
"""Return list of tool schemas."""
|
|
return [t.to_schema() for t in self._tools.values()]
|
|
|
|
def allowed_for(self, tool_name: str, agent_permissions: list[str] | str) -> bool:
|
|
"""Return True if agent is allowed to use this tool (* or matching scope)."""
|
|
tool = self._tools.get(tool_name)
|
|
if not tool:
|
|
return False
|
|
perms = agent_permissions if isinstance(agent_permissions, list) else [agent_permissions]
|
|
if "*" in tool.permission_scope or "*" in perms:
|
|
return True
|
|
return bool(set(tool.permission_scope) & set(perms))
|