FabledGrimoire/src/fabledassistant/services/tools/_registry.py

"""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
    journal: 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,
    journal: 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,
            journal=journal,
            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, *, conversation_type: str = "chat"
) -> list[dict]:
    """Build the tool schema list for a user, scoped by conversation type.

    - 'chat': all tools except those marked journal-only.
    - 'journal': all tools except set_rag_scope (scope is implicit in journal).
    """
    tools: list[dict] = []
    for td in _REGISTRY.values():
        if td.requires and not await _check_requires(user_id, td.requires):
            continue
        if conversation_type == "journal":
            if td.name == "set_rag_scope":
                continue
        else:
            if td.journal:
                continue
        tools.append(td.schema())
    logger.debug(
        "User %d / %s: %d tools available", user_id, conversation_type, len(tools)
    )
    return tools


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)}