"""MCP-Exposure for platform features (I-MCP). Exposes Search, Agents, Workflows, and Knowledge as thin MCP-compatible tools on top of existing tools/services. MCP possesses no own rights; the existing auth/run-as context and normal permission checks always apply. This is NOT a separate MCP server — it's a thin exposure layer that maps existing platform functions to MCP tool schemas so external MCP clients can invoke them. """ from __future__ import annotations import logging import uuid from typing import Any from sqlalchemy.ext.asyncio import AsyncSession logger = logging.getLogger(__name__) # ─── MCP Tool Definitions ──────────────────────────────────────────────────── MCP_TOOLS: list[dict[str, Any]] = [ { "name": "search", "description": "Search across all entities (contacts, companies, DMS, wiki, mail, communication).", "input_schema": { "type": "object", "properties": { "query": {"type": "string", "description": "Search query"}, "entity_type": {"type": "string", "description": "Optional entity type filter"}, "limit": {"type": "integer", "description": "Max results (default 10)", "default": 10}, }, "required": ["query"], }, "required_permission": "contacts:read", "handler": "search", }, { "name": "ask_knowledge", "description": "Query the knowledge base with RAG and evidence-backed results.", "input_schema": { "type": "object", "properties": { "query": {"type": "string", "description": "Natural language query"}, "source_types": { "type": "array", "items": {"type": "string"}, "description": "Optional source filter (wiki, dms, mail, communication)", }, "max_results": {"type": "integer", "description": "Max results (default 5)", "default": 5}, }, "required": ["query"], }, "required_permission": "contacts:read", "handler": "ask_knowledge", }, { "name": "start_workflow", "description": "Start a workflow instance by workflow ID.", "input_schema": { "type": "object", "properties": { "workflow_id": {"type": "string", "description": "Workflow definition ID"}, "context": {"type": "object", "description": "Initial context variables"}, }, "required": ["workflow_id"], }, "required_permission": "workflows:write", "handler": "start_workflow", }, { "name": "check_workflow_status", "description": "Check the status of a workflow instance.", "input_schema": { "type": "object", "properties": { "instance_id": {"type": "string", "description": "Workflow instance ID"}, }, "required": ["instance_id"], }, "required_permission": "workflows:read", "handler": "check_workflow_status", }, { "name": "list_agents", "description": "List available AI agents.", "input_schema": { "type": "object", "properties": {}, }, "required_permission": "agents:read", "handler": "list_agents", }, { "name": "create_task", "description": "Create a task (todo, follow-up, etc.).", "input_schema": { "type": "object", "properties": { "title": {"type": "string", "description": "Task title"}, "description": {"type": "string", "description": "Task description"}, "priority": {"type": "string", "description": "low|medium|high|urgent", "default": "medium"}, "entity_type": {"type": "string", "description": "Linked entity type"}, "entity_id": {"type": "string", "description": "Linked entity ID"}, }, "required": ["title"], }, "required_permission": "tasks:write", "handler": "create_task", }, ] def get_mcp_tools() -> list[dict[str, Any]]: """List all available MCP tools with their schemas.""" return [ { "name": t["name"], "description": t["description"], "input_schema": t["input_schema"], } for t in MCP_TOOLS ] def get_mcp_tool(name: str) -> dict[str, Any] | None: """Get a single MCP tool definition by name.""" return next((t for t in MCP_TOOLS if t["name"] == name), None) async def execute_mcp_tool( db: AsyncSession, tenant_id: uuid.UUID, user_id: uuid.UUID, tool_name: str, arguments: dict[str, Any], user_permissions: dict[str, Any] | None = None, ) -> dict[str, Any]: """Execute an MCP tool — thin wrapper over existing platform functions. MCP possesses no own rights. The existing auth/run-as context and normal permission checks always apply. This function checks the user's permissions before executing the tool. """ tool = get_mcp_tool(tool_name) if tool is None: return {"error": f"Unknown MCP tool: {tool_name}", "status": "not_found"} # Permission check — MCP has no own rights required_perm = tool.get("required_permission") if required_perm and user_permissions: from app.core.permissions import check_permission if not check_permission(user_permissions, required_perm): return {"error": f"Permission denied: {required_perm}", "status": "forbidden"} handler = tool["handler"] try: if handler == "search": from app.ai.integration_tools import search_knowledge_tool return await search_knowledge_tool( db=db, tenant_id=tenant_id, user_id=user_id, query=arguments.get("query", ""), entity_type=arguments.get("entity_type"), limit=arguments.get("limit", 10), ) elif handler == "ask_knowledge": from app.ai.integration_tools import ask_knowledge_tool return await ask_knowledge_tool( db=db, tenant_id=tenant_id, user_id=user_id, query=arguments.get("query", ""), source_types=arguments.get("source_types"), max_results=arguments.get("max_results", 5), ) elif handler == "start_workflow": from app.ai.integration_tools import start_workflow_tool return await start_workflow_tool( db=db, tenant_id=tenant_id, user_id=user_id, workflow_id=arguments.get("workflow_id", ""), context=arguments.get("context"), ) elif handler == "check_workflow_status": from app.ai.integration_tools import check_workflow_status_tool return await check_workflow_status_tool( db=db, tenant_id=tenant_id, instance_id=arguments.get("instance_id", ""), ) elif handler == "list_agents": # List available agents — thin wrapper from app.plugins.builtins.automation.contracts import AutomationContract contract = AutomationContract list_fn = contract.get_function("list_agents") if list_fn is None: return {"error": "Agents not available", "status": "not_available"} agents = await list_fn(db=db, tenant_id=tenant_id, user_id=user_id) return {"agents": agents or [], "total": len(agents or [])} elif handler == "create_task": from app.plugins.builtins.tasks.services import create_task result = await create_task( db=db, tenant_id=tenant_id, user_id=user_id, data=arguments, ) return result or {"error": "Failed to create task"} else: return {"error": f"Unknown handler: {handler}", "status": "not_implemented"} except Exception as e: logger.warning("MCP tool '%s' failed: %s", tool_name, e) return {"error": str(e), "status": "failed"} __all__ = [ "MCP_TOOLS", "get_mcp_tools", "get_mcp_tool", "execute_mcp_tool", ]