Agent Zero
46cadb5165
- Replace 9 hardcoded MCP tools with single call_crm_api tool - MCP clients now have full access to all 259 CRM API endpoints - Same generic approach as AI Assistant: method + path + body - Internal auth via X-Internal-Call headers with tenant/user context
127 lines
4.7 KiB
Python
127 lines
4.7 KiB
Python
"""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,
|
|
}
|