From 6ae0d5cbd1e9270d132ec73982ffbc66a2e6f365 Mon Sep 17 00:00:00 2001 From: Michael SCHAL Date: Sat, 18 Jul 2026 16:58:34 +0200 Subject: [PATCH] =?UTF-8?q?feat(mcp):=20gestionnaire=20de=20sessions=20laz?= =?UTF-8?q?y=20+=20conversion=20sch=C3=A9mas=20+=20appels?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Co-Authored-By: Claude Opus 4.8 --- backend/app/main.py | 4 +- backend/app/mcp_client.py | 195 ++++++++++++++++++++++++++++++ backend/pytest.ini | 2 + backend/tests/test_mcp_manager.py | 82 +++++++++++++ 4 files changed, 282 insertions(+), 1 deletion(-) create mode 100644 backend/pytest.ini create mode 100644 backend/tests/test_mcp_manager.py diff --git a/backend/app/main.py b/backend/app/main.py index 70ad050..51cdc20 100644 --- a/backend/app/main.py +++ b/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( diff --git a/backend/app/mcp_client.py b/backend/app/mcp_client.py index 5453476..2389a33 100644 --- a/backend/app/mcp_client.py +++ b/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 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() diff --git a/backend/pytest.ini b/backend/pytest.ini new file mode 100644 index 0000000..2f4c80e --- /dev/null +++ b/backend/pytest.ini @@ -0,0 +1,2 @@ +[pytest] +asyncio_mode = auto diff --git a/backend/tests/test_mcp_manager.py b/backend/tests/test_mcp_manager.py new file mode 100644 index 0000000..9a170ed --- /dev/null +++ b/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={})