bvandeusenandClaude Opus 4.6
ce2d76447c
Split 2566-line tools.py into a tools/ package with @tool decorator registration. Each tool's schema, metadata, and implementation live together. Briefing eligibility is now a briefing=True flag instead of a separate frozenset allowlist. Conditional inclusion (CalDAV, SearXNG) uses requires= metadata. Public API (get_tools_for_user, execute_tool) unchanged. Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
134 lines
4.0 KiB
Python
134 lines
4.0 KiB
Python
"""Decorator-based tool registry.
|
|
|
|
Each tool is registered via ``@tool(...)`` which captures the OpenAI-style
|
|
function schema and handler in one place. ``get_tools_for_user`` and
|
|
``execute_tool`` are the public API consumed by the rest of the app.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
from dataclasses import dataclass, field
|
|
from typing import Any, Callable, Coroutine
|
|
|
|
from fabledassistant.config import Config
|
|
from fabledassistant.services.caldav import is_caldav_configured
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
type ToolHandler = Callable[..., Coroutine[Any, Any, dict]]
|
|
|
|
|
|
@dataclass(frozen=True, slots=True)
|
|
class ToolDef:
|
|
name: str
|
|
description: str
|
|
parameters: dict
|
|
handler: ToolHandler
|
|
read_only: bool = False
|
|
briefing: bool = False
|
|
requires: str | None = None
|
|
required_params: list[str] = field(default_factory=list)
|
|
|
|
def schema(self) -> dict:
|
|
"""Return the OpenAI function-calling schema dict."""
|
|
return {
|
|
"type": "function",
|
|
"function": {
|
|
"name": self.name,
|
|
"description": self.description,
|
|
"parameters": {
|
|
"type": "object",
|
|
"properties": self.parameters,
|
|
**({"required": self.required_params} if self.required_params else {}),
|
|
},
|
|
},
|
|
}
|
|
|
|
|
|
_REGISTRY: dict[str, ToolDef] = {}
|
|
|
|
|
|
def tool(
|
|
*,
|
|
name: str,
|
|
description: str,
|
|
parameters: dict | None = None,
|
|
required: list[str] | None = None,
|
|
read_only: bool = False,
|
|
briefing: bool = False,
|
|
requires: str | None = None,
|
|
) -> Callable[[ToolHandler], ToolHandler]:
|
|
"""Register an async tool handler with its schema metadata."""
|
|
|
|
def decorator(fn: ToolHandler) -> ToolHandler:
|
|
if name in _REGISTRY:
|
|
raise ValueError(f"Duplicate tool registration: {name}")
|
|
_REGISTRY[name] = ToolDef(
|
|
name=name,
|
|
description=description,
|
|
parameters=parameters or {},
|
|
handler=fn,
|
|
read_only=read_only,
|
|
briefing=briefing,
|
|
requires=requires,
|
|
required_params=required or [],
|
|
)
|
|
return fn
|
|
|
|
return decorator
|
|
|
|
|
|
async def _check_requires(user_id: int, requires: str) -> bool:
|
|
if requires == "caldav":
|
|
return await is_caldav_configured(user_id)
|
|
if requires == "searxng":
|
|
return Config.searxng_enabled()
|
|
return True
|
|
|
|
|
|
async def get_tools_for_user(user_id: int) -> list[dict]:
|
|
"""Build the tool schema list for a user based on configured integrations."""
|
|
tools: list[dict] = []
|
|
for td in _REGISTRY.values():
|
|
if td.requires and not await _check_requires(user_id, td.requires):
|
|
continue
|
|
tools.append(td.schema())
|
|
logger.debug("User %d: %d tools available", user_id, len(tools))
|
|
return tools
|
|
|
|
|
|
async def get_briefing_tools(user_id: int) -> list[dict]:
|
|
"""Return only the tool schemas marked ``briefing=True``."""
|
|
all_tools = await get_tools_for_user(user_id)
|
|
names = {td.name for td in _REGISTRY.values() if td.briefing}
|
|
filtered = [t for t in all_tools if t["function"]["name"] in names]
|
|
logger.debug(
|
|
"Briefing tools for user %d: %d of %d selected",
|
|
user_id, len(filtered), len(all_tools),
|
|
)
|
|
return filtered
|
|
|
|
|
|
async def execute_tool(
|
|
user_id: int,
|
|
tool_name: str,
|
|
arguments: dict,
|
|
conv_id: int | None = None,
|
|
workspace_project_id: int | None = None,
|
|
) -> dict:
|
|
"""Execute a tool call and return the result."""
|
|
td = _REGISTRY.get(tool_name)
|
|
if td is None:
|
|
return {"success": False, "error": f"Unknown tool: {tool_name}"}
|
|
try:
|
|
return await td.handler(
|
|
user_id=user_id,
|
|
arguments=arguments,
|
|
conv_id=conv_id,
|
|
workspace_project_id=workspace_project_id,
|
|
)
|
|
except Exception as e:
|
|
logger.exception("Tool execution failed: %s", tool_name)
|
|
return {"success": False, "error": str(e)}
|