leocrm/app/ai/agent_tools.py

"""Tool-/Skill-Binding for AI agents.

Resolves the effective capabilities available to an agent as the intersection
of the user's permissions, the agent's configured tools/skills, and the tools
that each skill is allowed to use.

Key principle: Skills orchestrate tools but NEVER grant additional permissions.
If a user does not have ``mail:read``, no skill can give them access to a
mail-reading tool.
"""

from __future__ import annotations

import logging
from typing import Any

from app.ai.skill_registry import SkillDefinition, SkillRegistry

logger = logging.getLogger(__name__)


def _user_has_permission(
    user_permissions: dict[str, Any],
    required_permission: str | None,
) -> bool:
    """Check whether the user has the required permission for a tool.

    A tool without a required permission is always allowed. The check uses the
    same semantics as ``app.core.permissions.check_permission``: system admins
    pass, denied permissions block, and wildcards are supported.
    """
    if not required_permission:
        return True
    if user_permissions.get("is_system_admin"):
        return True

    denied = set(user_permissions.get("denied_permissions", []) or [])
    if any(_permission_matches(denied_perm, required_permission) for denied_perm in denied):
        return False

    granted = set(user_permissions.get("permissions", []) or [])
    return any(_permission_matches(granted_perm, required_permission) for granted_perm in granted)


def _permission_matches(granted: str, required: str) -> bool:
    """Match a granted permission against a required one, supporting wildcards."""
    if granted == required:
        return True
    g_parts = granted.split(":")
    r_parts = required.split(":")
    if len(g_parts) != len(r_parts):
        return False
    for g_part, r_part in zip(g_parts, r_parts, strict=False):
        if g_part == "*":
            continue
        if g_part != r_part:
            return False
    return True


def get_agent_tools(
    agent_definition: Any,
    tool_registry: Any,
    skill_registry: SkillRegistry,
    user_permissions: dict[str, Any],
) -> tuple[list[dict[str, Any]], list[SkillDefinition]]:
    """Get the tools and skills available to this agent.

    Effective capabilities = User/Run-as ∩ Agent ∩ Skill ∩ Tool.

    Args:
        agent_definition: AgentDefinition with ``tool_ids`` and ``skill_ids``.
        tool_registry: ToolRegistry with ``get_by_names`` and ``get``.
        skill_registry: SkillRegistry used to resolve skill names.
        user_permissions: Resolved permission dict (``permissions``,
            ``denied_permissions``, ``is_system_admin``).

    Returns:
        A tuple of (tool_schemas, skills). ``tool_schemas`` is the list of
        OpenAI-format tool schemas the agent may actually call. ``skills`` is
        the list of resolved SkillDefinitions the agent may use.
    """
    agent_tool_ids: list[str] = list(getattr(agent_definition, "tool_ids", None) or [])
    agent_skill_ids: list[str] = list(getattr(agent_definition, "skill_ids", None) or [])

    # 1. Resolve the agent's skills to SkillDefinitions.
    skills = skill_registry.get_by_names(agent_skill_ids)

    # 2. Collect the tool IDs available directly on the agent.
    direct_tool_ids = set(agent_tool_ids)

    # 3. For each skill, collect its allowed tool IDs.
    skill_tool_ids: set[str] = set()
    for skill in skills:
        skill_tool_ids.update(skill.allowed_tool_ids or [])

    # 4. Intersect: agent.tool_ids ∩ skill.allowed_tool_ids → tools via skills.
    #    Tools directly in agent.tool_ids (not via skills) are also available.
    available_tool_ids = direct_tool_ids | (direct_tool_ids & skill_tool_ids)

    # 5. Resolve the available tools from the registry.
    tools = tool_registry.get_by_names(sorted(available_tool_ids))

    # 6. Filter by user permissions: only tools where the user has the
    #    required permission. Skills never grant additional permissions.
    permitted_tools = [
        tool
        for tool in tools
        if _user_has_permission(user_permissions, getattr(tool, "required_permission", None))
    ]

    # 7. Build OpenAI-format schemas and return.
    tool_schemas = [tool.to_openai_schema() for tool in permitted_tools]
    return tool_schemas, skills


__all__ = ["get_agent_tools"]