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

"""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, get_current_user_or_bearer, 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(get_current_user_or_bearer),
) -> McpToolListResponse:
    """List all available MCP tools with their schemas.

    Accepts session cookie OR Bearer token.
    """
    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_or_bearer),
) -> McpToolExecuteResponse:
    """Execute an MCP tool by name with provided arguments.

    Accepts session cookie OR Bearer token (for programmatic access).
    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", []),
        "auth_method": current_user.get("_auth_method", "session"),
    }

    # Audit log
    from app.core.audit import log_audit
    import uuid as uuid_mod
    correlation_id = str(uuid_mod.uuid4())
    await log_audit(
        db,
        tenant_id=uuid.UUID(current_user["tenant_id"]),
        user_id=uuid.UUID(current_user["user_id"]),
        action="mcp.tool.execute",
        entity_type="mcp_tool",
        entity_id=None,
        details={"tool": tool_name, "arguments": request.arguments, "correlation_id": correlation_id, "auth_method": context["auth_method"]},
    )

    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(get_current_user_or_bearer),
) -> McpServerConfig:
    """Get MCP server configuration for external clients.

    Accepts session cookie OR Bearer token.
    """
    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(),
    )