feat: MCP Server - generic CRM API access instead of 9 hardcoded tools
- 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
This commit is contained in:
Agent Zero
@@ -1,18 +1,18 @@
|
||||
"""MCP tool definitions for LeoCRM — contacts, calendar, mail, DMS.
|
||||
"""MCP tool definitions for LeoCRM — generic CRM API access.
|
||||
|
||||
Each tool has a name, description, parameter schema, required permission,
|
||||
and an async handler that operates on the database.
|
||||
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 base64
|
||||
import json
|
||||
import logging
|
||||
import os
|
||||
import uuid
|
||||
from datetime import datetime
|
||||
from typing import Any
|
||||
|
||||
from sqlalchemy import select, and_
|
||||
import httpx
|
||||
from sqlalchemy.ext.asyncio import AsyncSession
|
||||
|
||||
from app.plugins.builtins.mcp_server.schemas import McpToolDefinition, McpToolParameter
|
||||
@@ -20,99 +20,31 @@ from app.plugins.builtins.mcp_server.schemas import McpToolDefinition, McpToolPa
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
# ─── Tool Definitions ──────────────────────────────────────────────────────
|
||||
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="search_contacts",
|
||||
description="Search contacts by name, email, or phone number.",
|
||||
category="contacts",
|
||||
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="query", type="string", description="Search query (name, email, phone)", required=True),
|
||||
McpToolParameter(name="limit", type="integer", description="Max results (default 20)", required=False, default=20),
|
||||
],
|
||||
),
|
||||
McpToolDefinition(
|
||||
name="get_contact",
|
||||
description="Get detailed information about a specific contact.",
|
||||
category="contacts",
|
||||
required_permission="mcp:read",
|
||||
parameters=[
|
||||
McpToolParameter(name="contact_id", type="string", description="Contact UUID", required=True),
|
||||
],
|
||||
),
|
||||
McpToolDefinition(
|
||||
name="create_contact",
|
||||
description="Create a new contact in LeoCRM.",
|
||||
category="contacts",
|
||||
required_permission="mcp:write",
|
||||
parameters=[
|
||||
McpToolParameter(name="name", type="string", description="Contact name", required=True),
|
||||
McpToolParameter(name="email", type="string", description="Contact email", required=False),
|
||||
McpToolParameter(name="phone", type="string", description="Contact phone", required=False),
|
||||
McpToolParameter(name="type", type="string", description="Contact type (person, company)", required=False, default="person"),
|
||||
],
|
||||
),
|
||||
McpToolDefinition(
|
||||
name="list_calendar_entries",
|
||||
description="List calendar entries within a date range.",
|
||||
category="calendar",
|
||||
required_permission="mcp:read",
|
||||
parameters=[
|
||||
McpToolParameter(name="date_from", type="string", description="Start date (ISO 8601)", required=False),
|
||||
McpToolParameter(name="date_to", type="string", description="End date (ISO 8601)", required=False),
|
||||
],
|
||||
),
|
||||
McpToolDefinition(
|
||||
name="create_calendar_entry",
|
||||
description="Create a new calendar appointment.",
|
||||
category="calendar",
|
||||
required_permission="mcp:write",
|
||||
parameters=[
|
||||
McpToolParameter(name="title", type="string", description="Appointment title", required=True),
|
||||
McpToolParameter(name="start", type="string", description="Start datetime (ISO 8601)", required=True),
|
||||
McpToolParameter(name="end", type="string", description="End datetime (ISO 8601)", required=False),
|
||||
],
|
||||
),
|
||||
McpToolDefinition(
|
||||
name="list_emails",
|
||||
description="List emails from a specific folder.",
|
||||
category="mail",
|
||||
required_permission="mcp:read",
|
||||
parameters=[
|
||||
McpToolParameter(name="folder", type="string", description="Folder name (default INBOX)", required=False, default="INBOX"),
|
||||
McpToolParameter(name="limit", type="integer", description="Max results (default 20)", required=False, default=20),
|
||||
],
|
||||
),
|
||||
McpToolDefinition(
|
||||
name="send_email",
|
||||
description="Send an email from LeoCRM.",
|
||||
category="mail",
|
||||
required_permission="mcp:write",
|
||||
parameters=[
|
||||
McpToolParameter(name="to", type="string", description="Recipient email address", required=True),
|
||||
McpToolParameter(name="subject", type="string", description="Email subject", required=True),
|
||||
McpToolParameter(name="body", type="string", description="Email body (plain text)", required=True),
|
||||
],
|
||||
),
|
||||
McpToolDefinition(
|
||||
name="list_files",
|
||||
description="List DMS files in a specific folder.",
|
||||
category="dms",
|
||||
required_permission="mcp:read",
|
||||
parameters=[
|
||||
McpToolParameter(name="folder_id", type="string", description="Folder UUID (optional, root if omitted)", required=False),
|
||||
],
|
||||
),
|
||||
McpToolDefinition(
|
||||
name="upload_file",
|
||||
description="Upload a file to the DMS.",
|
||||
category="dms",
|
||||
required_permission="mcp:write",
|
||||
parameters=[
|
||||
McpToolParameter(name="filename", type="string", description="File name", required=True),
|
||||
McpToolParameter(name="content_base64", type="string", description="File content as base64 string", required=True),
|
||||
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),
|
||||
],
|
||||
),
|
||||
]
|
||||
@@ -131,274 +63,64 @@ def get_all_tool_names() -> list[str]:
|
||||
return [t.name for t in TOOL_DEFINITIONS]
|
||||
|
||||
|
||||
# ─── Tool Handlers ──────────────────────────────────────────────────────────
|
||||
# ─── Tool Handler ─────────────────────────────────────────────────────────
|
||||
|
||||
|
||||
async def _handler_search_contacts(db: AsyncSession, arguments: dict[str, Any], context: dict[str, Any]) -> dict[str, Any]:
|
||||
"""Search contacts by name/email/phone."""
|
||||
from app.models.contact import Contact
|
||||
query = arguments.get("query", "")
|
||||
limit = min(int(arguments.get("limit", 20)), 100)
|
||||
pattern = f"%{query}%"
|
||||
stmt = select(Contact).where(
|
||||
and_(
|
||||
Contact.deleted_at.is_(None),
|
||||
(Contact.name.ilike(pattern))
|
||||
| (Contact.displayname.ilike(pattern))
|
||||
| (Contact.email_1.ilike(pattern))
|
||||
| (Contact.phone_1.ilike(pattern)),
|
||||
)
|
||||
).limit(limit)
|
||||
result = await db.execute(stmt)
|
||||
contacts = result.scalars().all()
|
||||
return {"contacts": [_contact_to_dict(c) for c in contacts], "count": len(contacts)}
|
||||
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
|
||||
|
||||
async def _handler_get_contact(db: AsyncSession, arguments: dict[str, Any], context: dict[str, Any]) -> dict[str, Any]:
|
||||
"""Get a single contact by ID."""
|
||||
from app.models.contact import Contact
|
||||
contact_id = arguments.get("contact_id")
|
||||
try:
|
||||
cid = uuid.UUID(contact_id)
|
||||
except (ValueError, TypeError):
|
||||
return {"error": "Invalid contact_id"}
|
||||
stmt = select(Contact).where(and_(Contact.id == cid, Contact.deleted_at.is_(None)))
|
||||
result = await db.execute(stmt)
|
||||
contact = result.scalar_one_or_none()
|
||||
if not contact:
|
||||
return {"error": "Contact not found"}
|
||||
return _contact_to_dict(contact)
|
||||
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)}
|
||||
|
||||
|
||||
async def _handler_create_contact(db: AsyncSession, arguments: dict[str, Any], context: dict[str, Any]) -> dict[str, Any]:
|
||||
"""Create a new contact."""
|
||||
from app.models.contact import Contact
|
||||
tenant_id = context.get("tenant_id")
|
||||
user_id = context.get("user_id")
|
||||
contact = Contact(
|
||||
tenant_id=uuid.UUID(tenant_id) if tenant_id else None,
|
||||
name=arguments.get("name", ""),
|
||||
displayname=arguments.get("name", ""),
|
||||
email_1=arguments.get("email"),
|
||||
phone_1=arguments.get("phone"),
|
||||
type=arguments.get("type", "person"),
|
||||
created_by=uuid.UUID(user_id) if user_id else None,
|
||||
)
|
||||
db.add(contact)
|
||||
await db.flush()
|
||||
return _contact_to_dict(contact)
|
||||
|
||||
|
||||
async def _handler_list_calendar_entries(db: AsyncSession, arguments: dict[str, Any], context: dict[str, Any]) -> dict[str, Any]:
|
||||
"""List calendar entries in a date range."""
|
||||
from app.plugins.builtins.calendar.models import CalendarEntry
|
||||
date_from = arguments.get("date_from")
|
||||
date_to = arguments.get("date_to")
|
||||
conditions = [CalendarEntry.deleted_at.is_(None)]
|
||||
if date_from:
|
||||
conditions.append(CalendarEntry.start_at >= datetime.fromisoformat(date_from))
|
||||
if date_to:
|
||||
conditions.append(CalendarEntry.start_at <= datetime.fromisoformat(date_to))
|
||||
stmt = select(CalendarEntry).where(and_(*conditions)).limit(50)
|
||||
result = await db.execute(stmt)
|
||||
entries = result.scalars().all()
|
||||
return {"entries": [_entry_to_dict(e) for e in entries], "count": len(entries)}
|
||||
|
||||
|
||||
async def _handler_create_calendar_entry(db: AsyncSession, arguments: dict[str, Any], context: dict[str, Any]) -> dict[str, Any]:
|
||||
"""Create a new calendar entry."""
|
||||
from app.plugins.builtins.calendar.models import Calendar, CalendarEntry
|
||||
tenant_id = context.get("tenant_id")
|
||||
user_id = context.get("user_id")
|
||||
# Find or create a default calendar
|
||||
cal_stmt = select(Calendar).where(and_(Calendar.tenant_id == uuid.UUID(tenant_id), Calendar.deleted_at.is_(None))).limit(1)
|
||||
cal_result = await db.execute(cal_stmt)
|
||||
calendar = cal_result.scalar_one_or_none()
|
||||
if not calendar:
|
||||
calendar = Calendar(
|
||||
tenant_id=uuid.UUID(tenant_id),
|
||||
name="MCP Default",
|
||||
owner_id=uuid.UUID(user_id),
|
||||
)
|
||||
db.add(calendar)
|
||||
await db.flush()
|
||||
entry = CalendarEntry(
|
||||
tenant_id=uuid.UUID(tenant_id),
|
||||
calendar_id=calendar.id,
|
||||
entry_type="appointment",
|
||||
title=arguments.get("title", ""),
|
||||
start_at=datetime.fromisoformat(arguments.get("start")) if arguments.get("start") else None,
|
||||
end_at=datetime.fromisoformat(arguments.get("end")) if arguments.get("end") else None,
|
||||
created_by=uuid.UUID(user_id),
|
||||
)
|
||||
db.add(entry)
|
||||
await db.flush()
|
||||
return _entry_to_dict(entry)
|
||||
|
||||
|
||||
async def _handler_list_emails(db: AsyncSession, arguments: dict[str, Any], context: dict[str, Any]) -> dict[str, Any]:
|
||||
"""List emails from a folder."""
|
||||
from app.plugins.builtins.mail.models import Mail, MailFolder
|
||||
folder_name = arguments.get("folder", "INBOX")
|
||||
limit = min(int(arguments.get("limit", 20)), 100)
|
||||
tenant_id = context.get("tenant_id")
|
||||
# Find folder by name
|
||||
folder_stmt = select(MailFolder).where(
|
||||
and_(MailFolder.tenant_id == uuid.UUID(tenant_id), MailFolder.name == folder_name)
|
||||
).limit(1)
|
||||
folder_result = await db.execute(folder_stmt)
|
||||
folder = folder_result.scalar_one_or_none()
|
||||
if not folder:
|
||||
return {"emails": [], "count": 0}
|
||||
mail_stmt = select(Mail).where(and_(Mail.folder_id == folder.id, Mail.deleted_at.is_(None))).limit(limit)
|
||||
mail_result = await db.execute(mail_stmt)
|
||||
mails = mail_result.scalars().all()
|
||||
return {"emails": [_mail_to_dict(m) for m in mails], "count": len(mails)}
|
||||
|
||||
|
||||
async def _handler_send_email(db: AsyncSession, arguments: dict[str, Any], context: dict[str, Any]) -> dict[str, Any]:
|
||||
"""Send an email — creates a mail record (actual SMTP handled by mail plugin)."""
|
||||
from app.plugins.builtins.mail.models import Mail, MailFolder
|
||||
tenant_id = context.get("tenant_id")
|
||||
user_id = context.get("user_id")
|
||||
# Find or create a Sent folder
|
||||
folder_stmt = select(MailFolder).where(
|
||||
and_(MailFolder.tenant_id == uuid.UUID(tenant_id), MailFolder.name == "Sent")
|
||||
).limit(1)
|
||||
folder_result = await db.execute(folder_stmt)
|
||||
folder = folder_result.scalar_one_or_none()
|
||||
if not folder:
|
||||
return {"error": "No Sent folder found — mail account not configured"}
|
||||
mail = Mail(
|
||||
tenant_id=uuid.UUID(tenant_id),
|
||||
folder_id=folder.id,
|
||||
from_addr="",
|
||||
to_addr=arguments.get("to", ""),
|
||||
subject=arguments.get("subject", ""),
|
||||
body_text=arguments.get("body", ""),
|
||||
is_outgoing=True,
|
||||
sent_at=datetime.utcnow(),
|
||||
)
|
||||
db.add(mail)
|
||||
await db.flush()
|
||||
return {"message": "Email queued for sending", "mail_id": str(mail.id)}
|
||||
|
||||
|
||||
async def _handler_list_files(db: AsyncSession, arguments: dict[str, Any], context: dict[str, Any]) -> dict[str, Any]:
|
||||
"""List DMS files in a folder."""
|
||||
from app.plugins.builtins.dms.models import File as DmsFile, Folder
|
||||
tenant_id = context.get("tenant_id")
|
||||
folder_id = arguments.get("folder_id")
|
||||
conditions = [DmsFile.tenant_id == uuid.UUID(tenant_id), DmsFile.deleted_at.is_(None)]
|
||||
if folder_id:
|
||||
conditions.append(DmsFile.folder_id == uuid.UUID(folder_id))
|
||||
stmt = select(DmsFile).where(and_(*conditions)).limit(50)
|
||||
result = await db.execute(stmt)
|
||||
files = result.scalars().all()
|
||||
return {"files": [_file_to_dict(f) for f in files], "count": len(files)}
|
||||
|
||||
|
||||
async def _handler_upload_file(db: AsyncSession, arguments: dict[str, Any], context: dict[str, Any]) -> dict[str, Any]:
|
||||
"""Upload a file to DMS (stores content in storage backend)."""
|
||||
from app.plugins.builtins.dms.models import File as DmsFile, Folder
|
||||
from app.core.storage import get_storage_backend
|
||||
tenant_id = context.get("tenant_id")
|
||||
user_id = context.get("user_id")
|
||||
filename = arguments.get("filename", "untitled")
|
||||
content_b64 = arguments.get("content_base64", "")
|
||||
try:
|
||||
content = base64.b64decode(content_b64)
|
||||
except Exception:
|
||||
return {"error": "Invalid base64 content"}
|
||||
# Find root folder
|
||||
folder_stmt = select(Folder).where(
|
||||
and_(Folder.tenant_id == uuid.UUID(tenant_id), Folder.parent_id.is_(None))
|
||||
).limit(1)
|
||||
folder_result = await db.execute(folder_stmt)
|
||||
folder = folder_result.scalar_one_or_none()
|
||||
if not folder:
|
||||
folder = Folder(tenant_id=uuid.UUID(tenant_id), name="Root", created_by=uuid.UUID(user_id))
|
||||
db.add(folder)
|
||||
await db.flush()
|
||||
# Store file
|
||||
storage = get_storage_backend()
|
||||
file_path = f"mcp/{uuid.uuid4()}/{filename}"
|
||||
storage.save(file_path, content)
|
||||
dms_file = DmsFile(
|
||||
tenant_id=uuid.UUID(tenant_id),
|
||||
folder_id=folder.id,
|
||||
name=filename,
|
||||
storage_path=file_path,
|
||||
mime_type="application/octet-stream",
|
||||
size=len(content),
|
||||
uploaded_by=uuid.UUID(user_id),
|
||||
)
|
||||
db.add(dms_file)
|
||||
await db.flush()
|
||||
return {"file_id": str(dms_file.id), "name": dms_file.name, "size": dms_file.size}
|
||||
|
||||
|
||||
# ─── Handler Registry ──────────────────────────────────────────────────────
|
||||
# ─── Handler Registry ─────────────────────────────────────────────────────
|
||||
|
||||
TOOL_HANDLERS: dict[str, Any] = {
|
||||
"search_contacts": _handler_search_contacts,
|
||||
"get_contact": _handler_get_contact,
|
||||
"create_contact": _handler_create_contact,
|
||||
"list_calendar_entries": _handler_list_calendar_entries,
|
||||
"create_calendar_entry": _handler_create_calendar_entry,
|
||||
"list_emails": _handler_list_emails,
|
||||
"send_email": _handler_send_email,
|
||||
"list_files": _handler_list_files,
|
||||
"upload_file": _handler_upload_file,
|
||||
"call_crm_api": _handler_call_crm_api,
|
||||
}
|
||||
|
||||
|
||||
# ─── Serialization Helpers ─────────────────────────────────────────────────
|
||||
|
||||
|
||||
def _contact_to_dict(c: Any) -> dict[str, Any]:
|
||||
return {
|
||||
"id": str(c.id),
|
||||
"name": c.name,
|
||||
"displayname": getattr(c, "displayname", c.name),
|
||||
"email": getattr(c, "email_1", None),
|
||||
"phone": getattr(c, "phone_1", None),
|
||||
"type": getattr(c, "type", "person"),
|
||||
"created_at": c.created_at.isoformat() if c.created_at else None,
|
||||
}
|
||||
|
||||
|
||||
def _entry_to_dict(e: Any) -> dict[str, Any]:
|
||||
return {
|
||||
"id": str(e.id),
|
||||
"calendar_id": str(e.calendar_id),
|
||||
"entry_type": e.entry_type,
|
||||
"title": e.title,
|
||||
"description": e.description,
|
||||
"start_at": e.start_at.isoformat() if e.start_at else None,
|
||||
"end_at": e.end_at.isoformat() if e.end_at else None,
|
||||
"status": e.status,
|
||||
}
|
||||
|
||||
|
||||
def _mail_to_dict(m: Any) -> dict[str, Any]:
|
||||
return {
|
||||
"id": str(m.id),
|
||||
"from": m.from_addr,
|
||||
"to": m.to_addr,
|
||||
"subject": m.subject,
|
||||
"body_preview": (m.body_text or "")[:200],
|
||||
"is_read": m.is_read,
|
||||
"sent_at": m.sent_at.isoformat() if m.sent_at else None,
|
||||
}
|
||||
|
||||
|
||||
def _file_to_dict(f: Any) -> dict[str, Any]:
|
||||
return {
|
||||
"id": str(f.id),
|
||||
"name": f.name,
|
||||
"size": f.size,
|
||||
"mime_type": f.mime_type,
|
||||
"folder_id": str(f.folder_id) if f.folder_id else None,
|
||||
"created_at": f.created_at.isoformat() if f.created_at else None,
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user