leocrm/app/plugins/builtins/ai_assistant/crm_api_tool.py

"""Generic CRM API tool — gives AI direct access to all CRM REST endpoints.

Instead of registering dozens of individual tools, this single tool lets the AI
call any CRM API endpoint. The OpenAPI spec is injected into the system prompt
so the AI knows which endpoints exist and what parameters they accept.
"""

from __future__ import annotations

import json
import logging
from typing import Any

import httpx

logger = logging.getLogger(__name__)

# Cache for OpenAPI spec
_openapi_cache: dict[str, Any] | None = None


def _get_base_url() -> str:
    """Get the internal base URL for API calls.

    F09 (Astra P1): inside the API container this is 127.0.0.1; the
    WORKER must reach the API service instead - 127.0.0.1 there points
    at the worker itself. INTERNAL_API_URL overrides (compose sets it
    to http://crm_app:PORT for the worker service).
    """
    import os

    override = os.environ.get("INTERNAL_API_URL")
    if override:
        return override.rstrip("/")
    port = os.environ.get("PORT", "8000")
    return f"http://127.0.0.1:{port}"


async def _make_internal_api_request(method, path, tenant_id, user_id, body=None):
    """F09 (Astra P1): authenticated internal API request.

    Sends a short-lived HMAC-signed delegation token (max 60 s) instead
    of the previous unauthenticated X-Internal-Call headers that the
    protected API never accepted. The token acts ON BEHALF OF the user
    - their real permissions apply (RBAC + RLS), no special rights.
    """
    from app.core.delegation_token import create_delegation_token

    token = create_delegation_token(
        user_id=str(user_id),
        tenant_id=str(tenant_id),
        agent_id="crm-api-tool",
    )
    headers = {
        "Content-Type": "application/json",
        "X-Delegation-Token": token,
    }
    url = f"{_get_base_url()}{path}"
    async with httpx.AsyncClient() as client:
        if method == "GET":
            return await client.get(url, headers=headers, timeout=30.0)
        if method == "POST":
            return await client.post(url, headers=headers, json=body, timeout=30.0)
        if method == "PATCH":
            return await client.patch(url, headers=headers, json=body, timeout=30.0)
        if method == "PUT":
            return await client.put(url, headers=headers, json=body, timeout=30.0)
        if method == "DELETE":
            return await client.delete(url, headers=headers, timeout=30.0)
        raise ValueError(f"Unsupported method: {method}")


async def get_openapi_spec() -> dict[str, Any]:
    """Get the CRM OpenAPI spec, cached."""
    global _openapi_cache
    if _openapi_cache is not None:
        return _openapi_cache

    try:
        base_url = _get_base_url()
        async with httpx.AsyncClient() as client:
            resp = await client.get(f"{base_url}/openapi.json", timeout=10.0)
            if resp.status_code == 200:
                spec = resp.json()
                # Compact the spec: only keep paths, methods, summary, parameters
                compact: dict[str, Any] = {"paths": {}}
                for path, methods in spec.get("paths", {}).items():
                    compact["paths"][path] = {}
                    for method, details in methods.items():
                        if method in ("get", "post", "patch", "put", "delete"):
                            compact["paths"][path][method] = {
                                "summary": details.get("summary", ""),
                                "description": details.get("description", "")[:200],
                                "parameters": [
                                    {
                                        "name": p.get("name", ""),
                                        "in": p.get("in", ""),
                                        "required": p.get("required", False),
                                        "schema": p.get("schema", {}),
                                    }
                                    for p in details.get("parameters", [])
                                ],
                                "requestBody": details.get("requestBody", {}).get("content", {}),
                            }
                _openapi_cache = compact
                return _openapi_cache
    except Exception as e:
        logger.warning("Failed to fetch OpenAPI spec: %s", e)

    _openapi_cache = {"paths": {}}
    return _openapi_cache


async def get_api_context_for_prompt() -> str:
    """Get a compact text summary of available API endpoints for the system prompt."""
    spec = await get_openapi_spec()
    lines: list[str] = ["\n\n## Verfügbare CRM API Endpunkte\n"]
    lines.append("Du kannst jeden dieser Endpunkte mit dem Tool 'call_crm_api' aufrufen.")
    lines.append("Parameter: method (GET/POST/PATCH/DELETE), path (z.B. /api/v1/contacts), body (JSON für POST/PATCH).\n")

    for path, methods in spec.get("paths", {}).items():
        for method, details in methods.items():
            summary = details.get("summary", "")
            lines.append(f"- {method.upper()} {path}: {summary}")

    return "\n".join(lines)


async def call_crm_api_handler(arguments: dict[str, Any], context: dict[str, Any]) -> str:
    """Execute an arbitrary CRM API call on behalf of the AI.

    The AI provides method, path, and optional body. The handler makes an
    internal HTTP request with the user's session context (tenant_id, user_id)
    and returns the response.
    """
    method = arguments.get("method", "GET").upper()
    path = arguments.get("path", "")
    body = arguments.get("body")

    if not path:
        return json.dumps({"error": "path is required"})

    # Ensure path starts with /
    if not path.startswith("/"):
        path = "/" + path

    try:
        # Get user context for auth
        tenant_id = context.get("tenant_id", "")
        user_id = context.get("user_id", "")

        # F09 (Astra P1): authenticated request via short-lived delegation
        # token - the acting user's real permissions apply.
        resp = await _make_internal_api_request(
            method, path, tenant_id=tenant_id, user_id=user_id, body=body
        )

        # Return response body (truncated if too large)
        try:
            resp_data = resp.json()
            resp_text = json.dumps(resp_data, default=str)
            if len(resp_text) > 8000:
                resp_text = resp_text[:8000] + "\n... [truncated]"
            return resp_text
        except Exception:
            return resp.text[:8000]

    except Exception as e:
        logger.exception("call_crm_api failed")
        return json.dumps({"error": str(e)})


# Tool definition for the registry
TOOL_NAME = "call_crm_api"
TOOL_DESCRIPTION = (
    "Rufe einen beliebigen CRM API Endpunkt auf. Verwende GET zum Lesen, "
    "POST zum Erstellen, PATCH zum Aktualisieren, DELETE zum Löschen. "
    "Der path muss mit /api/v1/ beginnen (z.B. /api/v1/contacts). "
    "Für POST/PATCH kann ein body (JSON Objekt) mitgegeben werden."
)
TOOL_PARAMETERS = {
    "type": "object",
    "properties": {
        "method": {
            "type": "string",
            "enum": ["GET", "POST", "PATCH", "PUT", "DELETE"],
            "default": "GET",
            "description": "HTTP Methode",
        },
        "path": {
            "type": "string",
            "description": "API Pfad, z.B. /api/v1/contacts oder /api/v1/contacts/{id}",
        },
        "body": {
            "type": "object",
            "description": "Request body für POST/PATCH (JSON Objekt)",
        },
    },
    "required": ["method", "path"],
}


def register_crm_api_tool(registry) -> None:
    """Register the generic CRM API tool."""
    registry.register(
        name=TOOL_NAME,
        description=TOOL_DESCRIPTION,
        parameters=TOOL_PARAMETERS,
        handler=call_crm_api_handler,
        plugin_name="ai_assistant",
        required_permission='ai:write',  # Uses internal auth context
        category="system",
    )
    logger.info("CRM API tool registered — AI can now call any CRM endpoint")