leocrm/app/plugins/builtins/unified_search/ai_tool.py

"""AI tool definition for Unified Search.

Exposes the unified search engine as a callable tool for AI agents.
The tool respects the calling user's permissions (tenant, visibility, RBAC)
by running the search with the user's context.
"""

from __future__ import annotations

import json
import logging
import uuid
from typing import Any

from app.plugins.builtins.unified_search.query_understanding import llm_analyze_query
from app.plugins.builtins.unified_search.search_engine import hybrid_search

logger = logging.getLogger(__name__)

TOOL_NAME = "unified_search"
TOOL_DESCRIPTION = (
    "Durchsuche alle CRM-Daten (Kontakte, Firmen, Mails, Dateien, Kalender, Tasks) "
    "mit Hybrid-Suche (Volltext + semantisch). "
    "Liefert kompakte Ergebnisse mit entity_type, title, snippet und score. "
    "Die Suche respektiert die Berechtigungen des aufrufenden Benutzers."
)
TOOL_PARAMETERS = {
    "type": "object",
    "properties": {
        "query": {
            "type": "string",
            "description": "Suchanfrage, z.B. 'Max Mustermann' oder 'Angebot 2026'",
        },
        "entity_types": {
            "type": "array",
            "items": {"type": "string"},
            "description": "Optional: Nur diese Entity-Typen durchsuchen (contact, company, mail, file, event, task, ...)",
        },
        "limit": {
            "type": "integer",
            "default": 10,
            "minimum": 1,
            "maximum": 50,
            "description": "Maximale Anzahl Ergebnisse (Standard: 10)",
        },
    },
    "required": ["query"],
}


async def unified_search_handler(arguments: dict[str, Any], context: dict[str, Any]) -> str:
    """Execute a unified search on behalf of the calling AI agent.

    Uses the user context (tenant_id, user_id, is_system_admin) from the AI
    session so that visibility filtering and RBAC are respected.
    """
    query = (arguments.get("query") or "").strip()
    if not query:
        return json.dumps({"error": "query is required"})

    entity_types = arguments.get("entity_types")
    if isinstance(entity_types, str):
        entity_types = [t.strip() for t in entity_types.split(",") if t.strip()]
    limit = int(arguments.get("limit", 10) or 10)
    limit = max(1, min(limit, 50))

    tenant_id = context.get("tenant_id")
    user_id = context.get("user_id")
    is_system_admin = bool(context.get("is_system_admin", False))

    if not tenant_id:
        return json.dumps({"error": "missing tenant context"})

    try:
        tenant_uuid = uuid.UUID(str(tenant_id))
        user_uuid = uuid.UUID(str(user_id)) if user_id else None
    except (ValueError, TypeError):
        return json.dumps({"error": "invalid tenant/user context"})

    # Build a DB session from the session factory (same pattern as other tools)
    from app.core.db import get_session_factory

    factory = get_session_factory()
    async with factory() as db:
        query_analysis = await llm_analyze_query(query, db=db, tenant_id=tenant_uuid)
        results = await hybrid_search(
            db=db,
            query_analysis=query_analysis,
            tenant_id=tenant_uuid,
            entity_types=entity_types,
            limit=limit,
            user_id=user_uuid,
            is_system_admin=is_system_admin,
        )

    # Compact AI-friendly output
    compact = [
        {
            "entity_type": r.get("entity_type", ""),
            "entity_id": r.get("entity_id", ""),
            "title": r.get("title", ""),
            "snippet": (r.get("snippet", "") or "")[:200],
            "score": round(float(r.get("score", 0.0)), 4),
        }
        for r in results
    ]

    return json.dumps({"count": len(compact), "results": compact}, ensure_ascii=False)


# Expose the tool definition object for direct import in verification
class _UnifiedSearchTool:
    """Lightweight tool descriptor matching the verification contract."""

    name = TOOL_NAME
    description = TOOL_DESCRIPTION
    parameters = TOOL_PARAMETERS
    handler = unified_search_handler
    plugin_name = "unified_search"
    required_permission = "search:read"
    category = "search"

    def to_openai_schema(self) -> dict[str, Any]:
        return {
            "type": "function",
            "function": {
                "name": self.name,
                "description": self.description,
                "parameters": self.parameters,
            },
        }


unified_search_tool = _UnifiedSearchTool()


def register_unified_search_tool(registry) -> None:
    """Register the unified_search tool in the AI tool registry."""
    registry.register(
        name=TOOL_NAME,
        description=TOOL_DESCRIPTION,
        parameters=TOOL_PARAMETERS,
        handler=unified_search_handler,
        plugin_name="unified_search",
        required_permission="search:read",
        category="search",
    )
    logger.info("Unified Search AI tool registered")