flowdeck/app/services/mcp_client.py

"""Client MCP (Model Context Protocol) — streamable HTTP, v7.57.0.

JSON-RPC 2.0 : ``initialize`` → ``notifications/initialized`` → ``tools/list``
→ ``tools/call``. Les outils sont **cachés** dans
``agent_connectors.tools_json`` (kind ``mcp``) puis exposés au LLM par
``ToolRegistry`` sous des noms ``mcp_<serveur>_<outil>`` — outils dynamiques,
sans état en mémoire.

ponytail : **1 seul seam** ``_rpc()`` (monkeypatché en test) ; on lit soit la
réponse JSON directe, soit la première ligne ``data:`` d'une
``text/event-stream`` — pas de client SSE à l'état (les méthodes utilisées
répondent en JSON chez la grande majorité des serveurs).
"""

from __future__ import annotations

import json
import logging
import re

from app.db import get_conn
from app.services.connectors import _headers, _with_secret, get

logger = logging.getLogger(__name__)

# Version la plus répandue côté serveurs (un serveur plus récent rétrograde).
PROTOCOL_VERSION = "2024-11-05"
CLIENT_INFO = {"name": "FlowDeck", "version": "7"}


def tool_name(server_name: str, tool_name: str) -> str:
    """Nom LLM stable : ``mcp_<serveur>_<outil>`` (slug minuscules/underscores)."""
    def slug(text: str) -> str:
        return re.sub(r"[^a-z0-9]+", "_", str(text).lower()).strip("_")
    joined = f"mcp_{slug(server_name)}_{slug(tool_name)}"
    return re.sub(r"_{2,}", "_", joined)


def parse_body(content_type: str, body: str, status: int) -> dict:
    """Corps MCP → dict JSON (JSON direct ou événement ``data:`` d'un flux SSE)."""
    text = body or ""
    if "text/event-stream" in (content_type or ""):
        for line in text.splitlines():
            if line.startswith("data:"):
                text = line[5:].strip()
                break
    try:
        data = json.loads(text)
    except ValueError as exc:
        raise ValueError(f"Réponse MCP illisible (HTTP {status})") from exc
    if isinstance(data, dict) and data.get("error"):
        err = data["error"] if isinstance(data["error"], dict) else {}
        raise ValueError(f"MCP {err.get('code', '?')} : {err.get('message', 'erreur')}")
    return data if isinstance(data, dict) else {}


async def _rpc(url: str, payload: dict, headers: dict | None = None) -> dict:
    """POST JSON-RPC gardé SSRF → réponse décodée. Point d'injection des tests."""
    from app.services.http_client import shared_client
    from app.services.importers.url_fetch import _validate_url

    safe = _validate_url(url)
    hdrs = {"Content-Type": "application/json",
            "Accept": "application/json, text/event-stream",
            **(headers or {})}
    async with shared_client(timeout=15, headers=hdrs) as client:
        resp = await client.post(safe, json=payload)
        return parse_body(resp.headers.get("content-type", ""), resp.text or "",
                          resp.status_code)


async def _notify(url: str, payload: dict, headers: dict | None = None) -> None:
    """Notification JSON-RPC (202 sans corps) — les erreurs sont ignorées."""
    try:
        await _rpc(url, payload, headers)
    except Exception as exc:  # noqa: BLE001
        logger.debug("MCP notification ignorée: %s", exc)


def _server(connector_id: int) -> dict:
    row = get(connector_id)
    if row is None:
        raise ValueError("Serveur MCP introuvable")
    if row["kind"] != "mcp":
        raise ValueError("Ce connecteur n'est pas un serveur MCP")
    if not row["enabled"]:
        raise ValueError("Serveur MCP désactivé")
    return {"id": row["id"], "name": row["name"], "url": row["url"],
            "headers": _headers(connector_id)}


def _normalize(server: dict, raw: list) -> list[dict]:
    out = []
    for t in raw if isinstance(raw, list) else []:
        if not isinstance(t, dict) or not t.get("name"):
            continue
        params = t.get("inputSchema")
        if not isinstance(params, dict):
            params = {"type": "object", "properties": {}}
        out.append({
            "name": tool_name(server["name"], t["name"]),
            "original": str(t["name"]),
            "description": str(t.get("description") or "Outil MCP"),
            "parameters": params,
        })
    return out


def _save_tools(connector_id: int, tools: list[dict]) -> None:
    with get_conn() as conn:
        conn.execute("UPDATE agent_connectors SET tools_json=? WHERE id=?",
                     (json.dumps(tools), connector_id))
        conn.commit()


async def initialize_and_list(connector_id: int) -> list[dict]:
    """Handshake + ``tools/list`` → met à jour le cache de la base."""
    srv = _server(connector_id)
    url = _with_secret(srv["url"], connector_id)
    await _rpc(url, {
        "jsonrpc": "2.0", "id": 1, "method": "initialize",
        "params": {"protocolVersion": PROTOCOL_VERSION, "capabilities": {},
                   "clientInfo": CLIENT_INFO},
    }, srv["headers"])
    await _notify(url, {"jsonrpc": "2.0", "method": "notifications/initialized"},
                  srv["headers"])
    data = await _rpc(url, {"jsonrpc": "2.0", "id": 2, "method": "tools/list",
                            "params": {}}, srv["headers"])
    result = data.get("result") if isinstance(data.get("result"), dict) else {}
    tools = _normalize(srv, result.get("tools") or [])
    _save_tools(connector_id, tools)
    return tools


def cached_tools() -> list[dict]:
    """Outils MCP des serveurs **activés** (cache en base) — lu par ToolRegistry."""
    with get_conn() as conn:
        rows = conn.execute(
            "SELECT id, tools_json FROM agent_connectors WHERE kind='mcp' AND enabled=1"
        ).fetchall()
    out: list[dict] = []
    for r in rows:
        try:
            tools = json.loads(r["tools_json"] or "[]")
        except (ValueError, TypeError):
            tools = []
        for t in tools if isinstance(tools, list) else []:
            if isinstance(t, dict) and t.get("name"):
                out.append({**t, "connector_id": r["id"]})
    return out


def tools_text(connector_id: int) -> str:
    """Liste lisible des outils cachés (utilisée par ``connector_fetch``)."""
    tools = [t for t in cached_tools() if t["connector_id"] == connector_id]
    if not tools:
        return "Aucun outil en cache — lancez « Tester le connecteur » (tools/list)."
    return "\n".join(f"- {t['name']} : {t.get('description', '')}" for t in tools)


def _content_text(result: dict) -> str:
    parts = []
    for item in result.get("content") or []:
        if isinstance(item, dict) and item.get("type") == "text":
            parts.append(str(item.get("text") or ""))
        else:
            parts.append(json.dumps(item, ensure_ascii=False)[:2000])
    return "\n".join(p for p in parts if p) or json.dumps(result, ensure_ascii=False)[:4000]


async def call_tool(connector_id: int, tool: str, arguments: dict | None = None) -> dict:
    """``tools/call`` → {status, text} (isError → error, borné à 20 000 car.)."""
    srv = _server(connector_id)
    url = _with_secret(srv["url"], connector_id)
    data = await _rpc(url, {
        "jsonrpc": "2.0", "id": 3, "method": "tools/call",
        "params": {"name": tool, "arguments": arguments or {}},
    }, srv["headers"])
    result = data.get("result") if isinstance(data.get("result"), dict) else {}
    text = _content_text(result)[:20000]
    return {"status": "error" if result.get("isError") else "ok", "text": text}