api/mcp.py

"""
backend/api/mcp.py — P19-B3: MCP (Model Context Protocol) server skeleton.

Implementa JSON-RPC 2.0 / MCP spec 2024-11-05.
Endpoint: POST /api/mcp

Compatibile con: Claude Desktop, Cursor, cline, Continue.dev, qualsiasi client MCP.
Auth: Bearer token via MCP_API_KEY env var (P20-Q1). Se non impostato → accesso libero (dev).

Metodi supportati:
  initialize    → handshake + capabilities negotiation
  tools/list    → lista dei 5 tool principali con JSON Schema input
  tools/call    → esegui tool per nome con arguments + timeout 30s

Design:
  - Tool whitelist esplicita (_MCP_TOOLS) — zero dipendenza da struttura interna TOOL_REGISTRY.
  - Fail-open: ogni errore tool → MCP error response, mai exception non gestita.
  - Batch support: body può essere array di richieste (MCP spec §4.3).
  - Timeout 30s per tool call — non blocca il worker FastAPI.
"""
from __future__ import annotations

import asyncio
import json
import logging
import os
import time
from typing import Any
import uuid

from fastapi import APIRouter, Request
from fastapi.responses import JSONResponse, StreamingResponse

router = APIRouter(tags=["mcp"])
_logger = logging.getLogger("agente_ai.mcp")

# P20-Q1: Auth — se MCP_API_KEY è impostato nell'env, richiede Authorization: Bearer <key>.
# In dev locale senza env var → accesso libero (utile per test locali con Claude Desktop).
# In prod (Railway): impostare MCP_API_KEY nei Railway env vars.
_MCP_API_KEY: str = os.getenv("MCP_API_KEY", "").strip()

_MCP_PROTOCOL_VERSION = "2024-11-05"
_SERVER_INFO = {"name": "agente-ai", "version": "1.0.0"}

# ── Tool whitelist MCP (MVP — top-5 strumenti principali) ───────────────────────
# Campo "inputSchema" = JSON Schema standard (MCP spec §5.1.1)

_MCP_TOOLS: list[dict] = [
    {
        "name":        "web_search",
        "description": (
            "Cerca informazioni aggiornate sul web tramite Brave, Tavily, Wikipedia, DuckDuckGo. "
            "Restituisce titoli, snippet e URL dei risultati più rilevanti."
        ),
        "inputSchema": {
            "type": "object",
            "properties": {
                "query":       {"type": "string",  "description": "Query di ricerca in linguaggio naturale"},
                "max_results": {"type": "integer", "description": "Numero massimo risultati (default 5)", "default": 5},
            },
            "required": ["query"],
        },
    },
    {
        "name":        "trigger_webhook",
        "description": (
            "Invia una richiesta HTTP POST/GET/PUT a un URL esterno (n8n, Pipedream, Zapier, "
            "Discord, Slack, CI/CD). Restituisce status HTTP e body risposta."
        ),
        "inputSchema": {
            "type": "object",
            "properties": {
                "url":     {"type": "string",  "description": "URL destinazione (https://...)"},
                "payload": {"type": "object",  "description": "Payload JSON da inviare"},
                "method":  {"type": "string",  "enum": ["POST", "GET", "PUT"], "default": "POST"},
                "headers": {"type": "object",  "description": "Header HTTP aggiuntivi"},
                "timeout": {"type": "number",  "description": "Timeout secondi (max 10)", "default": 10},
            },
            "required": ["url"],
        },
    },
    {
        "name":        "memory_read",
        "description": (
            "Legge dalla memoria episodica dell'agente: recupera ricordi semanticamente simili "
            "alla query tramite embedding vector search (top-k risultati)."
        ),
        "inputSchema": {
            "type": "object",
            "properties": {
                "query":   {"type": "string",  "description": "Query per la ricerca semantica"},
                "top_k":   {"type": "integer", "description": "Numero max di ricordi restituiti (default 5)", "default": 5},
            },
            "required": ["query"],
        },
    },
    {
        "name":        "code_exec",
        "description": (
            "Esegue codice Python in una sandbox isolata (backend-exec microservice). "
            "Restituisce stdout, stderr e exit code. Timeout 25s."
        ),
        "inputSchema": {
            "type": "object",
            "properties": {
                "code":    {"type": "string",  "description": "Codice Python da eseguire"},
                "timeout": {"type": "number",  "description": "Timeout in secondi (max 25)", "default": 20},
            },
            "required": ["code"],
        },
    },
    {
        "name":        "file_write",
        "description": (
            "Scrive o aggiorna un file nel workspace virtuale (VFS) dell'agente. "
            "Percorso relativo alla root del progetto corrente."
        ),
        "inputSchema": {
            "type": "object",
            "properties": {
                "path":    {"type": "string",  "description": "Percorso file (es. 'src/index.ts')"},
                "content": {"type": "string",  "description": "Contenuto completo del file"},
            },
            "required": ["path", "content"],
        },
    },
]

_MCP_TOOL_MAP: dict[str, dict] = {t["name"]: t for t in _MCP_TOOLS}


# ── Tool dispatch ───────────────────────────────────────────────────────────────

async def _dispatch_tool(name: str, arguments: dict) -> Any:
    """
    Chiama il tool reale con gli argomenti MCP. Timeout 30s.
    Ogni tool ha il proprio import lazy per evitare circular dependency.
    """
    if name == "web_search":
        from tools.registry import _web_search as _ws
        return await asyncio.wait_for(_ws(**arguments), timeout=30.0)

    if name == "trigger_webhook":
        from tools.trigger_webhook import trigger_webhook as _tw
        return await asyncio.wait_for(_tw(**arguments), timeout=30.0)

    if name == "memory_read":
        # Accede al MemoryManager singleton via api.state
        from api.state import _get_mem_manager_async as _gmm
        mem = await _gmm()
        if mem is None:
            return {"error": "memory_manager non disponibile"}
        query   = arguments.get("query", "")
        top_k   = int(arguments.get("top_k", 5))
        results = await asyncio.wait_for(mem.retrieve(query, top_k=top_k), timeout=10.0)
        return [{"content": r.get("content", ""), "score": r.get("score", 0.0)} for r in results]

    if name == "code_exec":
        from tools.registry import _call_exec_engine as _cee
        code    = arguments.get("code", "")
        timeout = float(arguments.get("timeout", 20))
        result  = await asyncio.wait_for(
            _cee({"code": code, "timeout": timeout}),
            timeout=timeout + 5,
        )
        return result or {"error": "exec engine non disponibile"}

    if name == "file_write":
        # Scrive su VFS Supabase — proxy via api.files
        try:
            from api.files import _write_file_internal as _wfi
            path    = arguments.get("path", "")
            content = arguments.get("content", "")
            await asyncio.wait_for(_wfi(path, content), timeout=10.0)
            return {"ok": True, "path": path, "bytes": len(content)}
        except (ImportError, AttributeError):
            return {"error": "file_write non disponibile — implementare _write_file_internal in api.files"}

    raise ValueError(f"Tool '{name}' non implementato nel dispatcher MCP")


# ── JSON-RPC helpers ────────────────────────────────────────────────────────────

def _ok(id_: Any, result: Any) -> dict:
    return {"jsonrpc": "2.0", "id": id_, "result": result}

def _err(id_: Any, code: int, message: str, data: Any = None) -> dict:
    e: dict[str, Any] = {"code": code, "message": message}
    if data is not None:
        e["data"] = str(data)[:500]
    return {"jsonrpc": "2.0", "id": id_, "error": e}


async def _handle_one(req: Any) -> dict:
    """Gestisce una singola richiesta JSON-RPC 2.0."""
    if not isinstance(req, dict):
        return _err(None, -32600, "Invalid Request")
    req_id = req.get("id")
    method = req.get("method", "")
    params = req.get("params") or {}

    if req.get("jsonrpc") != "2.0":
        return _err(req_id, -32600, "jsonrpc deve essere '2.0'")

    # ── initialize ───────────────────────────────────────────────────────────
    if method == "initialize":
        client_info = params.get("clientInfo", {})
        _logger.info("[mcp] initialize — client: %s %s",
                     client_info.get("name", "unknown"),
                     client_info.get("version", ""))
        return _ok(req_id, {
            "protocolVersion": _MCP_PROTOCOL_VERSION,
            "capabilities":    {"tools": {"listChanged": False}},
            "serverInfo":      _SERVER_INFO,
        })

    # ── tools/list ───────────────────────────────────────────────────────────
    if method == "tools/list":
        return _ok(req_id, {"tools": _MCP_TOOLS})

    # ── tools/call ───────────────────────────────────────────────────────────
    if method == "tools/call":
        tool_name = params.get("name")
        arguments = params.get("arguments") or {}
        if not tool_name:
            return _err(req_id, -32602, "Params mancante: 'name' obbligatorio")
        if tool_name not in _MCP_TOOL_MAP:
            return _err(req_id, -32602,
                        f"Tool '{tool_name}' non in whitelist MCP",
                        data={"available": list(_MCP_TOOL_MAP.keys())})
        try:
            t0 = time.monotonic()
            result = await _dispatch_tool(tool_name, arguments)
            elapsed = int((time.monotonic() - t0) * 1000)
            _logger.info("[mcp] tools/call %s OK (%dms)", tool_name, elapsed)

            # Normalizza → MCP content[] format
            if isinstance(result, str):
                content = [{"type": "text", "text": result}]
            elif isinstance(result, (dict, list)):
                content = [{"type": "text",
                            "text": json.dumps(result, ensure_ascii=False, indent=2)}]
            else:
                content = [{"type": "text", "text": str(result)}]

            return _ok(req_id, {
                "content": content,
                "isError": False,
                "_meta":   {"elapsed_ms": elapsed},
            })
        except asyncio.TimeoutError:
            _logger.warning("[mcp] tools/call %s — timeout 30s", tool_name)
            return _err(req_id, -32603, f"Tool '{tool_name}' timeout (30s)")
        except Exception as exc:
            _logger.warning("[mcp] tools/call %s — errore: %s", tool_name, exc)
            return _ok(req_id, {
                "content": [{"type": "text", "text": f"Errore: {exc!s:.500}"}],
                "isError": True,
            })

    # ── notifications (fire-and-forget — no response per spec) ───────────────
    if method.startswith("notifications/"):
        _logger.debug("[mcp] notification ricevuta: %s", method)
        return {}   # spec: notification non ha risposta

    # ── Method not found ──────────────────────────────────────────────────────
    return _err(req_id, -32601, f"Method not found: {method!r}")


# ── Main endpoint ───────────────────────────────────────────────────────────────

@router.post("/api/mcp")
async def mcp_endpoint(request: Request) -> JSONResponse:
    """
    POST /api/mcp — MCP JSON-RPC 2.0 endpoint.

    Compatibile con Claude Desktop (mcp://), Cursor, Continue.dev.
    Supporta sia request singola che batch (array).
    Auth: se MCP_API_KEY è configurato, richiede `Authorization: Bearer <key>`.
    """
    # ── P20-Q1: Auth gate ────────────────────────────────────────────────────
    if _MCP_API_KEY:
        auth_header = request.headers.get("authorization", "")
        if not auth_header.lower().startswith("bearer "):
            return JSONResponse(
                _err(None, -32000, "Unauthorized: Authorization Bearer header mancante"),
                status_code=401,
                headers={"WWW-Authenticate": "Bearer realm=\"agente-ai-mcp\""},
            )
        provided_key = auth_header[7:].strip()
        if provided_key != _MCP_API_KEY:
            _logger.warning("[mcp] auth fallita — chiave errata")
            return JSONResponse(
                _err(None, -32000, "Unauthorized: MCP API key non valida"),
                status_code=401,
                headers={"WWW-Authenticate": "Bearer realm=\"agente-ai-mcp\""},
            )

    try:
        body = await request.json()
    except Exception:
        return JSONResponse(
            _err(None, -32700, "Parse error: body JSON non valido"),
            status_code=400,
        )

    if isinstance(body, list):
        # Batch request — processa tutte in parallelo
        results = await asyncio.gather(*[_handle_one(r) for r in body])
        # Filtra notification responses (dict vuoto)
        out = [r for r in results if r]
        return JSONResponse(out)

    result = await _handle_one(body)
    if not result:   # notification — nessuna risposta
        return JSONResponse(None, status_code=204)
    return JSONResponse(result)


# ── P20-B2: SSE transport (MCP spec 2024-11-05 §transport-sse) ────────────────
# Permette ai client MCP che richiedono SSE (es. Claude Desktop, alcuni proxy)
# di aprire uno stream persistente invece del request/response classico.
#
# Flow:
#   1. Client → GET /api/mcp/sse          → riceve "event: endpoint" con POST URL
#   2. Client → POST /api/mcp/messages?sessionId=<id>  → invia JSON-RPC
#   3. Server → invia risposta JSON-RPC come "data:" event sul SSE stream
#
# Session store: dizionario in-memory (asyncio.Queue per sessione).
# TTL: 1h implicito — la queue viene rimossa quando il client chiude lo stream.

_SSE_SESSIONS: dict[str, asyncio.Queue] = {}


def _sse_auth_check(request: "Request") -> bool:
    """Ritorna True se autorizzato (o MCP_API_KEY non configurata)."""
    if not _MCP_API_KEY:
        return True
    auth = request.headers.get("authorization", "")
    if not auth.lower().startswith("bearer "):
        return False
    return auth[7:].strip() == _MCP_API_KEY


@router.get("/api/mcp/sse")
async def mcp_sse_endpoint(request: Request) -> StreamingResponse:
    """
    GET /api/mcp/sse — MCP SSE transport session init (P20-B2).

    Apre uno stream SSE e invia immediatamente l'endpoint per i messaggi.
    Mantiene la connessione viva con keepalive ogni 30s.
    Client chiude → session rimossa automaticamente.
    """
    if not _sse_auth_check(request):
        return JSONResponse(  # type: ignore[return-value]
            {"error": "Unauthorized", "hint": "Authorization: Bearer <MCP_API_KEY>"},
            status_code=401,
            headers={"WWW-Authenticate": "Bearer realm=\"agente-ai-mcp\""},
        )

    session_id: str    = str(uuid.uuid4())
    queue: asyncio.Queue = asyncio.Queue()
    _SSE_SESSIONS[session_id] = queue

    async def event_stream():
        try:
            # 1. Notifica endpoint per i messaggi
            endpoint_url = f"/api/mcp/messages?sessionId={session_id}"
            yield f"event: endpoint\ndata: {endpoint_url}\n\n"

            # 2. Stream risposta + keepalive
            while True:
                try:
                    msg = await asyncio.wait_for(queue.get(), timeout=30.0)
                    if msg is None:          # sentinel → chiudi stream
                        break
                    yield f"data: {json.dumps(msg, ensure_ascii=False)}\n\n"
                except asyncio.TimeoutError:
                    yield ": keepalive\n\n"   # comment line — nessun evento
        except asyncio.CancelledError:
            pass  # client disconnesso
        finally:
            _SSE_SESSIONS.pop(session_id, None)

    return StreamingResponse(
        event_stream(),
        media_type="text/event-stream",
        headers={
            "Cache-Control": "no-cache",
            "X-Accel-Buffering": "no",   # disabilita buffering Nginx
            "Connection":       "keep-alive",
        },
    )


@router.post("/api/mcp/messages")
async def mcp_messages_endpoint(request: Request, sessionId: str) -> JSONResponse:
    """
    POST /api/mcp/messages?sessionId=<id> — MCP SSE messages endpoint (P20-B2).

    Riceve JSON-RPC (singolo o batch), processa con _handle_one(),
    invia il risultato alla queue SSE della sessione, risponde 202 Accepted.
    Il client riceve la risposta reale come SSE event sul canale aperto.
    """
    if not _sse_auth_check(request):
        return JSONResponse(
            {"error": "Unauthorized"},
            status_code=401,
            headers={"WWW-Authenticate": "Bearer realm=\"agente-ai-mcp\""},
        )

    queue = _SSE_SESSIONS.get(sessionId)
    if queue is None:
        return JSONResponse(
            {"error": "session_not_found", "sessionId": sessionId,
             "hint": "Apri prima GET /api/mcp/sse per ottenere il sessionId"},
            status_code=404,
        )

    try:
        body = await request.json()
    except Exception:
        return JSONResponse({"error": "invalid_json"}, status_code=400)

    # Processa con la stessa logica del POST /api/mcp
    if isinstance(body, list):
        results: Any = await asyncio.gather(*[_handle_one(r) for r in body])
        result_payload = [r for r in results if r]  # filtra notification
    else:
        result_payload = await _handle_one(body)

    # Invia risposta alla SSE queue
    if result_payload:
        await queue.put(result_payload)

    return JSONResponse({"ok": True, "sessionId": sessionId}, status_code=202)