feat(mcp): gestionnaire de sessions lazy + conversion schémas + appels

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

backend/app/main.py

@@ -49,8 +49,10 @@ async def lifespan(_: FastAPI):
     # Préchargement du modèle par défaut, sans bloquer le démarrage.
     asyncio.create_task(_warm_default_model())
     yield
-    # Ferme le pool HTTP partagé vers Ollama.
+    # Ferme le pool HTTP partagé vers Ollama et les sessions MCP.
     await ollama.aclose()
+    from .mcp_client import manager as mcp_manager
+    await mcp_manager.aclose()
 
 
 app = FastAPI(

backend/app/mcp_client.py

@@ -6,7 +6,13 @@ session réutilisée ensuite. Toute panne est non bloquante pour le chat.
 """
 from __future__ import annotations
 
+import asyncio
 import logging
+import shlex
+from contextlib import AsyncExitStack
+
+from mcp import ClientSession, StdioServerParameters
+from mcp.client.stdio import stdio_client
 
 from . import db
 
@@ -87,3 +93,192 @@ def set_mcp_state(sid: str, *, enabled: bool, params: dict) -> dict:
     state[sid] = {"enabled": enabled, "params": dict(params)}
     db.set_config_value(MCP_KEY, state)
     return state
+
+
+_CALL_TIMEOUT = 30.0
+_CONNECT_TIMEOUT = 20.0
+_MAX_RESULT_CHARS = 8000
+
+
+class _ServerConn:
+    """Session vivante vers un serveur MCP (process stdio + handshake)."""
+
+    def __init__(self, sid: str) -> None:
+        self.sid = sid
+        self.stack = AsyncExitStack()
+        self.session: ClientSession | None = None
+        self.tools: list[dict] = []  # définitions format Ollama
+
+    async def start(self) -> None:
+        entry = CATALOG[self.sid]
+        params = get_mcp_state()[self.sid]["params"]
+        spec = params.get("command", "").strip()
+        if entry["command"] is None and spec.startswith(("http://", "https://")):
+            # Serveur "custom" en streamable HTTP (URL collée par l'utilisateur).
+            from mcp.client.streamable_http import streamablehttp_client
+            read, write, _ = await self.stack.enter_async_context(
+                streamablehttp_client(spec)
+            )
+        else:
+            if entry["command"] is None:
+                # Serveur "custom" : commande stdio saisie par l'utilisateur.
+                command = shlex.split(spec)
+                if not command:
+                    raise ValueError("commande du serveur personnalisé vide")
+            else:
+                command = list(entry["command"])
+            env = {k: params[k] for k in entry["env_params"] if params.get(k)}
+            server = StdioServerParameters(
+                command=command[0], args=command[1:], env=env or None
+            )
+            read, write = await self.stack.enter_async_context(
+                stdio_client(server)
+            )
+        self.session = await self.stack.enter_async_context(
+            ClientSession(read, write)
+        )
+        await asyncio.wait_for(self.session.initialize(), _CONNECT_TIMEOUT)
+        listed = await asyncio.wait_for(self.session.list_tools(), _CONNECT_TIMEOUT)
+        expose = entry.get("expose")
+        self.tools = [
+            {
+                "type": "function",
+                "function": {
+                    "name": f"mcp_{self.sid}_{t.name}",
+                    "description": (t.description or t.name)[:400],
+                    "parameters": t.inputSchema
+                    or {"type": "object", "properties": {}},
+                },
+            }
+            for t in listed.tools
+            if expose is None or t.name in expose
+        ]
+
+    async def close(self) -> None:
+        try:
+            await self.stack.aclose()
+        except Exception:  # process déjà mort : sans importance
+            pass
+
+
+class McpManager:
+    """Sessions MCP lazy + dispatch d'appels d'outils, jamais bloquant."""
+
+    def __init__(self) -> None:
+        self._conns: dict[str, _ServerConn] = {}
+        self._errors: dict[str, str] = {}
+        self._notices: list[str] = []
+        self._lock = asyncio.Lock()
+
+    async def _ensure(self, sid: str) -> _ServerConn | None:
+        if sid in self._conns:
+            return self._conns[sid]
+        try:
+            conn = _ServerConn(sid)
+            await conn.start()
+        except Exception as exc:
+            msg = str(exc)[:300] or exc.__class__.__name__
+            logger.warning("Serveur MCP %s indisponible : %s", sid, msg)
+            self._errors[sid] = msg
+            self._notices.append(
+                f"Serveur MCP « {CATALOG[sid]['label']} » indisponible : {msg}"
+            )
+            return None
+        self._errors.pop(sid, None)
+        self._conns[sid] = conn
+        return conn
+
+    async def tool_definitions(self) -> list[dict]:
+        """Outils des serveurs activés (connexion lazy, pannes ignorées)."""
+        defs: list[dict] = []
+        async with self._lock:
+            state = get_mcp_state()
+            # Ferme les serveurs désactivés entre-temps.
+            for sid in [s for s in self._conns if not state[s]["enabled"]]:
+                await self._conns.pop(sid).close()
+            for sid, st in state.items():
+                if not st["enabled"]:
+                    continue
+                conn = await self._ensure(sid)
+                if conn:
+                    defs.extend(conn.tools)
+        return defs
+
+    async def call_tool(self, prefixed_name: str, args: dict) -> dict:
+        # mcp_<sid>_<tool> ; sid ne contient pas de "_", le nom d'outil peut.
+        try:
+            _, sid, tool = prefixed_name.split("_", 2)
+        except ValueError:
+            return {"ok": False, "content": "", "summary": "nom d'outil invalide"}
+        async with self._lock:
+            conn = self._conns.get(sid) or await self._ensure(sid)
+        if conn is None or conn.session is None:
+            return {"ok": False, "content": "",
+                    "summary": f"serveur MCP {sid} indisponible"}
+        try:
+            result = await asyncio.wait_for(
+                conn.session.call_tool(tool, args or {}), _CALL_TIMEOUT
+            )
+        except Exception as exc:
+            # Session probablement morte : on la ferme, retry au prochain tour.
+            async with self._lock:
+                dead = self._conns.pop(sid, None)
+            if dead:
+                await dead.close()
+            return {"ok": False, "content": "",
+                    "summary": f"échec MCP : {str(exc)[:200]}"}
+        parts = [
+            c.text for c in result.content
+            if getattr(c, "type", "") == "text" and getattr(c, "text", "")
+        ]
+        content = "\n".join(parts)[:_MAX_RESULT_CHARS]
+        ok = not bool(getattr(result, "isError", False))
+        return {
+            "ok": ok,
+            "content": content,
+            "summary": (content.splitlines()[0][:120] if content else "terminé")
+            if ok else (content[:120] or "erreur outil MCP"),
+        }
+
+    def statuses(self) -> dict[str, dict]:
+        state = get_mcp_state()
+        out = {}
+        for sid, st in state.items():
+            if sid in self._conns:
+                s = "connected"
+            elif sid in self._errors:
+                s = "error"
+            else:
+                s = "inactive"
+            out[sid] = {
+                "state": s if st["enabled"] else "inactive",
+                "error": self._errors.get(sid),
+                "tools": len(self._conns[sid].tools) if sid in self._conns else 0,
+            }
+        return out
+
+    def notices(self) -> list[str]:
+        out, self._notices = self._notices, []
+        return out
+
+    async def test_server(self, sid: str) -> dict:
+        """Connexion d'essai indépendante (n'altère pas les sessions)."""
+        conn = _ServerConn(sid)
+        try:
+            await conn.start()
+            return {"ok": True,
+                    "tools": [d["function"]["name"] for d in conn.tools],
+                    "error": None}
+        except Exception as exc:
+            return {"ok": False, "tools": [],
+                    "error": str(exc)[:300] or exc.__class__.__name__}
+        finally:
+            await conn.close()
+
+    async def aclose(self) -> None:
+        for conn in list(self._conns.values()):
+            await conn.close()
+        self._conns.clear()
+
+
+manager = McpManager()

backend/pytest.ini

@@ -0,0 +1,2 @@
+[pytest]
+asyncio_mode = auto

backend/tests/test_mcp_manager.py

@@ -0,0 +1,82 @@
+import os
+import sys
+import tempfile
+import textwrap
+
+os.environ.setdefault("DATA_DIR", tempfile.mkdtemp())
+os.environ.setdefault("WORKSPACE_DIR", tempfile.mkdtemp())
+
+import pytest  # noqa: E402
+
+from app import db, mcp_client  # noqa: E402
+
+db.init_db()
+
+# Serveur MCP minimal : un outil "echo" qui renvoie son argument.
+_FAKE_SERVER = textwrap.dedent("""
+    from mcp.server.fastmcp import FastMCP
+    mcp = FastMCP("fake")
+
+    @mcp.tool()
+    def echo(text: str) -> str:
+        \"\"\"Répète le texte fourni.\"\"\"
+        return "echo:" + text
+
+    mcp.run()
+""")
+
+
+@pytest.fixture()
+def fake_server_cmd(tmp_path):
+    path = tmp_path / "fake_mcp.py"
+    path.write_text(_FAKE_SERVER, encoding="utf-8")
+    return [sys.executable, str(path)]
+
+
+@pytest.mark.asyncio
+async def test_tools_exposed_and_called(fake_server_cmd, monkeypatch):
+    monkeypatch.setitem(
+        mcp_client.CATALOG, "fake",
+        {"label": "Fake", "description": "", "command": fake_server_cmd,
+         "url_param": False, "env_params": [], "expose": None},
+    )
+    mcp_client.set_mcp_state("fake", enabled=True, params={})
+    mgr = mcp_client.McpManager()
+    try:
+        defs = await mgr.tool_definitions()
+        names = [d["function"]["name"] for d in defs]
+        assert "mcp_fake_echo" in names
+        result = await mgr.call_tool("mcp_fake_echo", {"text": "bonjour"})
+        assert result["ok"] is True
+        assert "echo:bonjour" in result["content"]
+    finally:
+        await mgr.aclose()
+        mcp_client.set_mcp_state("fake", enabled=False, params={})
+
+
+@pytest.mark.asyncio
+async def test_disabled_server_exposes_nothing():
+    mgr = mcp_client.McpManager()
+    try:
+        assert await mgr.tool_definitions() == []
+    finally:
+        await mgr.aclose()
+
+
+@pytest.mark.asyncio
+async def test_broken_server_never_raises(monkeypatch):
+    monkeypatch.setitem(
+        mcp_client.CATALOG, "broken",
+        {"label": "Broken", "description": "",
+         "command": [sys.executable, "-c", "import sys; sys.exit(3)"],
+         "url_param": False, "env_params": [], "expose": None},
+    )
+    mcp_client.set_mcp_state("broken", enabled=True, params={})
+    mgr = mcp_client.McpManager()
+    try:
+        assert await mgr.tool_definitions() == []
+        assert mgr.statuses()["broken"]["state"] == "error"
+        assert any("broken" in n.lower() or "Broken" in n for n in mgr.notices())
+    finally:
+        await mgr.aclose()
+        mcp_client.set_mcp_state("broken", enabled=False, params={})