leocrm/app/plugins/builtins/mcp_server/tool_definitions.py

"""MCP tool definitions for LeoCRM — generic CRM API access.

Single tool that gives MCP clients (Claude Desktop, etc.) full access to all
CRM API endpoints via the generic call_crm_api tool.
"""

from __future__ import annotations

import json
import logging
import os
import uuid
from typing import Any

import httpx
from sqlalchemy.ext.asyncio import AsyncSession

from app.plugins.builtins.mcp_server.schemas import McpToolDefinition, McpToolParameter

logger = logging.getLogger(__name__)


def _get_base_url() -> str:
    """Get the internal base URL for API calls."""
    port = os.environ.get("PORT", "8000")
    return f"http://127.0.0.1:{port}"


# ─── Generic CRM API Tool ──────────────────────────────────────────────────

TOOL_DEFINITIONS: list[McpToolDefinition] = [
    McpToolDefinition(
        name="call_crm_api",
        description=(
            "Rufe einen beliebigen CRM API Endpunkt auf. "
            "GET zum Lesen, POST zum Erstellen, PATCH zum Aktualisieren, DELETE zum Löschen. "
            "Verfügbare Endpunkte: /api/v1/contacts, /api/v1/mail, /api/v1/calendar, "
            "/api/v1/dms, /api/v1/tasks, /api/v1/addresses, /api/v1/bank-accounts, "
            "/api/v1/search, /api/v1/comm, /api/v1/ai, /api/v1/reports, etc. "
            "Für POST/PATCH kann ein body (JSON) mitgegeben werden."
        ),
        category="system",
        required_permission="mcp:read",
        parameters=[
            McpToolParameter(name="method", type="string", description="HTTP Methode: GET, POST, PATCH, DELETE", required=True),
            McpToolParameter(name="path", type="string", description="API Pfad, z.B. /api/v1/contacts", required=True),
            McpToolParameter(name="body", type="object", description="Request body für POST/PATCH (JSON Objekt)", required=False),
        ],
    ),
]


def get_tool_definition(name: str) -> McpToolDefinition | None:
    """Get a tool definition by name."""
    for tool in TOOL_DEFINITIONS:
        if tool.name == name:
            return tool
    return None


def get_all_tool_names() -> list[str]:
    """Get all tool names."""
    return [t.name for t in TOOL_DEFINITIONS]


# ─── Tool Handler ─────────────────────────────────────────────────────────


async def _handler_call_crm_api(db: AsyncSession, arguments: dict[str, Any], context: dict[str, Any]) -> dict[str, Any]:
    """Execute an arbitrary CRM API call on behalf of the MCP client."""
    method = arguments.get("method", "GET").upper()
    path = arguments.get("path", "")
    body = arguments.get("body")

    if not path:
        return {"error": "path is required"}

    if not path.startswith("/"):
        path = "/" + path

    try:
        base_url = _get_base_url()
        tenant_id = context.get("tenant_id", "")
        user_id = context.get("user_id", "")

        headers = {
            "Content-Type": "application/json",
            "X-Internal-Call": "true",
            "X-Tenant-Id": str(tenant_id),
            "X-User-Id": str(user_id),
        }

        async with httpx.AsyncClient() as client:
            if method == "GET":
                resp = await client.get(f"{base_url}{path}", headers=headers, timeout=30.0)
            elif method == "POST":
                resp = await client.post(f"{base_url}{path}", headers=headers, json=body, timeout=30.0)
            elif method == "PATCH":
                resp = await client.patch(f"{base_url}{path}", headers=headers, json=body, timeout=30.0)
            elif method == "PUT":
                resp = await client.put(f"{base_url}{path}", headers=headers, json=body, timeout=30.0)
            elif method == "DELETE":
                resp = await client.delete(f"{base_url}{path}", headers=headers, timeout=30.0)
            else:
                return {"error": f"Unsupported method: {method}"}

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

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


# ─── Handler Registry ─────────────────────────────────────────────────────

TOOL_HANDLERS: dict[str, Any] = {
    "call_crm_api": _handler_call_crm_api,
}