feat: MCP Server plugin (Task 5.16) - expose LeoCRM tools to external AI clients

- New plugin app/plugins/builtins/mcp_server/ with 9 MCP tools
- Tools: search_contacts, get_contact, create_contact, list_calendar_entries,
  create_calendar_entry, list_emails, send_email, list_files, upload_file
- Routes: GET /api/v1/mcp/tools, POST /api/v1/mcp/tools/{name}/execute, GET /api/v1/mcp/config
- API-token auth via session + RBAC (mcp:read, mcp:write)
- Frontend: mcp.ts API client with React Query hooks
- Frontend: SettingsMcp.tsx settings page with tool listing and execution
- i18n: de.json and en.json updated with MCP entries
- Tests: 7 tests covering tool listing, config, execution, auth, schema validation
This commit is contained in:
Agent Zero
2026-07-23 23:01:59 +02:00
parent f4beb78f91
commit 9d4f701a25
14 changed files with 1313 additions and 6 deletions
+129
View File
@@ -0,0 +1,129 @@
"""MCP Server plugin routes — tool listing, execution, and config."""
from __future__ import annotations
import logging
import uuid
from typing import Any
from fastapi import APIRouter, Depends, Header, HTTPException, status
from sqlalchemy.ext.asyncio import AsyncSession
from app.core.db import get_db
from app.deps import get_current_user, require_permission
from app.plugins.builtins.mcp_server.schemas import (
McpServerConfig,
McpToolDefinition,
McpToolExecuteRequest,
McpToolExecuteResponse,
McpToolListResponse,
)
from app.plugins.builtins.mcp_server.tool_definitions import (
TOOL_DEFINITIONS,
TOOL_HANDLERS,
get_all_tool_names,
get_tool_definition,
)
logger = logging.getLogger(__name__)
router = APIRouter(prefix="/api/v1/mcp", tags=["mcp-server"])
async def _get_mcp_context(
current_user: dict[str, Any] = Depends(get_current_user),
) -> dict[str, Any]:
"""Build MCP execution context from authenticated user."""
return {
"tenant_id": current_user.get("tenant_id"),
"user_id": current_user.get("user_id"),
"role": current_user.get("role"),
"permissions": current_user.get("permissions", []),
}
@router.get("/tools", response_model=McpToolListResponse)
async def list_mcp_tools(
current_user: dict[str, Any] = Depends(require_permission("mcp:read")),
) -> McpToolListResponse:
"""List all available MCP tools with their schemas."""
return McpToolListResponse(
tools=TOOL_DEFINITIONS,
count=len(TOOL_DEFINITIONS),
)
@router.post("/tools/{tool_name}/execute", response_model=McpToolExecuteResponse)
async def execute_mcp_tool(
tool_name: str,
request: McpToolExecuteRequest,
db: AsyncSession = Depends(get_db),
current_user: dict[str, Any] = Depends(get_current_user),
) -> McpToolExecuteResponse:
"""Execute an MCP tool by name with provided arguments.
Requires mcp:read for read tools, mcp:write for write tools.
"""
tool_def = get_tool_definition(tool_name)
if not tool_def:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail={"detail": f"MCP tool '{tool_name}' not found", "code": "tool_not_found"},
)
# Check permission based on tool category
required_perm = tool_def.required_permission or "mcp:read"
from app.core.permissions import check_permission
if not current_user.get("is_system_admin") and not check_permission(current_user, required_perm):
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail={"detail": f"Permission '{required_perm}' required", "code": "forbidden"},
)
handler = TOOL_HANDLERS.get(tool_name)
if not handler:
raise HTTPException(
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
detail={"detail": f"Handler for '{tool_name}' not implemented", "code": "handler_missing"},
)
context = {
"tenant_id": current_user.get("tenant_id"),
"user_id": current_user.get("user_id"),
"role": current_user.get("role"),
"permissions": current_user.get("permissions", []),
}
try:
result = await handler(db, request.arguments, context)
await db.commit()
return McpToolExecuteResponse(
tool=tool_name,
success="error" not in result,
result=result,
error=result.get("error"),
)
except Exception as exc:
logger.exception("MCP tool execution failed: %s", tool_name)
await db.rollback()
return McpToolExecuteResponse(
tool=tool_name,
success=False,
error=str(exc),
)
@router.get("/config", response_model=McpServerConfig)
async def get_mcp_config(
current_user: dict[str, Any] = Depends(require_permission("mcp:read")),
) -> McpServerConfig:
"""Get MCP server configuration for external clients."""
return McpServerConfig(
server_name="LeoCRM",
server_version="1.0.0",
protocol_version="2024-11-05",
base_url="/api/v1/mcp",
auth_method="api-token",
available_tools=get_all_tool_names(),
)