feat(fusion): serveur MCP complété (ACC-5.1) + pont MCP ObsiGate (ACC-2.2) — 7.83.0

ACC-5.1 (#24) — app/routers/mcp_server.py : catalogue §7.2 complété
(list_projects, get_project, read_page tronqué proprement,
list_project_tasks), search_pages en plein texte, liens profonds +
kind sur chaque résultat, scopes du jeton appliqués (écriture
refusée -32002 sans scope write) et journalisation §7.4 de chaque
tools/call dans api_audit_log (clés d'arguments seulement).

ACC-2.2 (#20) — pont MCP Flowdeck → ObsiGate, côté client :
- mcp_client : serveurs à session (capture/réémission de
  Mcp-Session-Id, handshake avant tools/call), décodage SSE
  multi-messages par id de réponse, classification des outils à la
  découverte (readOnlyHint sinon heuristique ; lecture par défaut).
- preset connecteur « obsigate » (URL dérivée dev/prod), kinds MCP
  factorisés (connectors.MCP_KINDS), option dans le panneau agent.
- garde-fou n° 2 : permission_manager.is_write_tool couvre les
  outils MCP écrivants → portes politique d'approbation
  (agent_engine) et rôles (assert_can) ; McpTool bloque tout
  apply_* distant (patron §7.5).

Tests : test_fusion_mcp_server 18, nouveau test_fusion_pont_obsigate
7 (faux serveur ObsiGate à session stricte via le vrai _rpc).
Preuve réelle T0-T5 : attend la clé ObsiGate dédiée (1 vault) et
Q-A/Q-B côté Bruno.

app/services/agent_engine.py

@@ -22,7 +22,7 @@ from app.services import agent_memory
 from app.services.agent_policies import check_tool, get_policy
 from app.services.context_builder import ContextBuilder
 from app.services.llm_client import LLMClient
-from app.services.permission_manager import WRITE_TOOLS, PermissionManager
+from app.services.permission_manager import PermissionManager, is_write_tool
 from app.services.tool_registry import ToolRegistry
 
 logger = logging.getLogger(__name__)
@@ -248,7 +248,7 @@ class AgentEngine:
                     # v7.2.0 — workspace tool scope + human approval gate, checked
                     # *before* permissions so a scoped-out tool never reaches ACLs.
                     gov = check_tool(self.user_id, self.workspace_id, tool,
-                                     is_write=tool in WRITE_TOOLS,
+                                     is_write=is_write_tool(tool),
                                      conversation_id=conversation_id)
                     if not gov.get("allowed"):
                         detail = gov.get("reason") or "Refusé par la politique agent"

app/services/connectors.py

@@ -24,7 +24,11 @@ from app.services.sso_provisioning import decrypt_secret, encrypt_secret
 logger = logging.getLogger(__name__)
 
 NATIVE_KINDS = ("gitea", "github", "web", "google", "ms365")
-CUSTOM_KINDS = ("custom", "discord", "telegram", "mcp", "tiktak")
+CUSTOM_KINDS = ("custom", "discord", "telegram", "mcp", "tiktak", "obsigate")
+#: Kinds personnels qui parlent MCP (probe = handshake + tools/list,
+#: outils dynamiques ``mcp_<serveur>_*``) — générique + les deux presets
+#: de fédération du programme (TikTak ACC-2.1, ObsiGate ACC-2.2).
+MCP_KINDS = ("mcp", "tiktak", "obsigate")
 AUTH_SCHEMES = ("bearer", "bot", "none")
 # Presets proposés par le formulaire du menu ＋ (le jeton reste toujours saisi
 # par l'utilisateur — jamais codé en dur ici).
@@ -45,6 +49,16 @@ def _presets() -> dict:
         # v7.82.0 : URL dérivée de la config (TIKTAK_URL ; défaut officiel).
         "tiktak": {"url": f"{draco_urls.tiktak_url()}/mcp", "auth": "bearer",
                    "hint": "TikTak → Réglages → Intégrations → clé API tck_…"},
+        # Fusion ACC-2.2 — pont MCP vers ObsiGate (plan « étape 1 :
+        # lectures seules », tests T0–T5) : serveur MCP d'ObsiGate sur
+        # /mcp (Streamable HTTP à session — le client gère le
+        # Mcp-Session-Id). La clé se crée dans ObsiGate →
+        # Configurations → Clés API & MCP, dédiée au pont et limitée
+        # à UN SEUL vault au départ : c'est le jeton qui fait la
+        # portée (garde-fou n° 1 du plan), pas ce preset.
+        "obsigate": {"url": f"{draco_urls.obsigate_url()}/mcp", "auth": "bearer",
+                     "hint": "ObsiGate → Configurations → Clés API & MCP → "
+                             "clé dédiée au pont, limitée à un seul vault"},
     }
 
 
@@ -293,7 +307,7 @@ async def probe(connector: str, user_id: int | None = None) -> dict:
     """Teste un connecteur et **persiste** le résultat (personnel uniquement)."""
     kind, ref = _resolve(connector)
     try:
-        if kind in ("mcp", "tiktak"):
+        if kind in MCP_KINDS:
             from app.services import mcp_client
             tools = await mcp_client.initialize_and_list(int(ref))
             status, detail = "ok", f"MCP · {len(tools)} outil(s)"
@@ -343,7 +357,7 @@ async def connector_fetch(connector: str, path: str = "", query: str = "",
     path = (path or "").strip()
     if kind in OAUTH_KINDS:
         return await oauth_connectors.api_get(kind, user_id, path)
-    if kind in ("mcp", "tiktak"):
+    if kind in MCP_KINDS:
         from app.services import mcp_client
         return {"status": "ok", "text": mcp_client.tools_text(int(ref))}
     if kind == "gitea":

app/services/mcp_client.py

@@ -6,10 +6,30 @@ JSON-RPC 2.0 : ``initialize`` → ``notifications/initialized`` → ``tools/list
 ``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).
+**Sessions (ACC-2.2, pont ObsiGate — test T0 du plan d'étape 1).**
+Certains serveurs Streamable HTTP (ObsiGate) sont **à session** : ils
+renvoient un en-tête ``Mcp-Session-Id`` à ``initialize`` et refusent
+tout appel qui ne le réémet pas. Le client capture donc cet en-tête à
+chaque réponse et le réémet sur les appels suivants de la même
+opération ; ``call_tool`` fait désormais le handshake complet avant
+``tools/call`` (les serveurs sans état comme TikTak l'acceptent sans
+frais). Réponses **JSON ou SSE** : dans un flux ``text/event-stream``
+qui porte plusieurs messages (notifications + réponse), c'est le
+message **de réponse** (id correspondant, sinon porteur de
+``result``/``error``) qui est retenu — pas le premier ``data:`` venu.
+
+**Classification des outils (garde-fou n° 2 du plan du pont).** Chaque
+outil découvert est classé au moment du cache : **lecture par défaut** ;
+écrivant si le serveur le déclare (annotation ``readOnlyHint: false``)
+ou si son nom porte un verbe d'écriture (``create_``, ``propose_``,
+``apply_``…). La classe est consommée par les portes de gouvernance
+(``agent_policies.check_tool`` et ``PermissionManager`` via
+``permission_manager.is_write_tool``) : un outil MCP écrivant suit le
+même régime d'approbation et de rôles que les outils internes.
+
+ponytail : **1 seul seam** ``_rpc()`` (monkeypatché en test) ; pas de
+client SSE à l'état (les méthodes utilisées répondent en JSON ou en
+événements ponctuels chez les serveurs du programme).
 """
 
 from __future__ import annotations
@@ -19,7 +39,7 @@ import logging
 import re
 
 from app.db import get_conn
-from app.services.connectors import _headers, _with_secret, get
+from app.services.connectors import MCP_KINDS, _headers, _with_secret, get
 
 logger = logging.getLogger(__name__)
 
@@ -27,6 +47,36 @@ logger = logging.getLogger(__name__)
 PROTOCOL_VERSION = "2024-11-05"
 CLIENT_INFO = {"name": "FlowDeck", "version": "7"}
 
+#: Verbes d'écriture reconnus en tête d'un nom d'outil MCP quand le
+#: serveur ne déclare pas d'annotation ``readOnlyHint``. ``propose_`` /
+#: ``apply_`` sont le patron d'écriture approuvée d'ObsiGate : les deux
+#: sont classés écrivants (voir aussi le blocage d'``apply_`` dans
+#: ``tool_registry.McpTool``).
+_PREFIXES_ECRITURE = (
+    "create_", "update_", "delete_", "add_", "remove_", "set_", "move_",
+    "write_", "edit_", "rename_", "propose_", "apply_", "archive_",
+    "unarchive_", "send_", "publish_", "close_", "sync_", "trigger_",
+    "import_",
+)
+
+#: Derniers ``Mcp-Session-Id`` capturés, par URL de serveur. Mémoire de
+#: processus volontairement éphémère : une session perdue se rouvre par
+#: un handshake, jamais par une persistance.
+_sessions: dict[str, str] = {}
+
+
+def classer_outil(nom_original: str, annotations: dict | None = None) -> bool:
+    """True si l'outil MCP est **écrivant**.
+
+    L'annotation du serveur fait foi quand elle existe
+    (``readOnlyHint``) ; sinon, heuristique de nom. Lecture par défaut —
+    c'est la règle du plan du pont : un outil ne devient écrivant que
+    s'il est déclaré comme tel.
+    """
+    if isinstance(annotations, dict) and "readOnlyHint" in annotations:
+        return not bool(annotations["readOnlyHint"])
+    return (nom_original or "").lower().startswith(_PREFIXES_ECRITURE)
+
 
 def tool_name(server_name: str, tool_name: str) -> str:
     """Nom LLM stable : ``mcp_<serveur>_<outil>`` (slug minuscules/underscores)."""
@@ -36,24 +86,66 @@ def tool_name(server_name: str, tool_name: str) -> str:
     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)."""
+def parse_body(content_type: str, body: str, status: int,
+               expected_id=None) -> dict:
+    """Corps MCP → dict JSON (JSON direct ou événements ``data:`` SSE).
+
+    Dans un flux SSE multi-messages, le message retenu est celui dont
+    l'``id`` correspond à la requête (``expected_id``), sinon le
+    premier porteur de ``result``/``error``, sinon le premier message.
+    """
     text = body or ""
+    data = None
     if "text/event-stream" in (content_type or ""):
+        candidats: list = []
         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
+                try:
+                    candidats.append(json.loads(line[5:].strip()))
+                except ValueError:
+                    continue
+        if expected_id is not None:
+            data = next((c for c in candidats
+                         if isinstance(c, dict) and c.get("id") == expected_id),
+                        None)
+        if data is None:
+            data = next((c for c in candidats
+                         if isinstance(c, dict)
+                         and ("result" in c or "error" in c)), None)
+        if data is None and candidats:
+            data = candidats[0]
+        if data is None:
+            raise ValueError(f"Réponse MCP illisible (HTTP {status})")
+    else:
+        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 {}
 
 
+def _capturer_session(url: str, headers) -> None:
+    """Mémorise le ``Mcp-Session-Id`` d'une réponse, s'il est présent."""
+    try:
+        sid = headers.get("mcp-session-id") if headers else None
+    except AttributeError:
+        sid = None
+    if sid:
+        _sessions[url] = sid
+
+
+def _session_headers(url: str, headers: dict | None) -> dict:
+    """En-têtes d'appel + session connue pour cette URL (si capturée)."""
+    out = dict(headers or {})
+    sid = _sessions.get(url)
+    if sid and not any(k.lower() == "mcp-session-id" for k in out):
+        out["Mcp-Session-Id"] = sid
+    return out
+
+
 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
@@ -65,8 +157,9 @@ async def _rpc(url: str, payload: dict, headers: dict | None = None) -> dict:
             **(headers or {})}
     async with shared_client(timeout=15, headers=hdrs) as client:
         resp = await client.post(safe, json=payload)
+        _capturer_session(url, resp.headers)
         return parse_body(resp.headers.get("content-type", ""), resp.text or "",
-                          resp.status_code)
+                          resp.status_code, expected_id=payload.get("id"))
 
 
 async def _notify(url: str, payload: dict, headers: dict | None = None) -> None:
@@ -77,11 +170,30 @@ async def _notify(url: str, payload: dict, headers: dict | None = None) -> None:
         logger.debug("MCP notification ignorée: %s", exc)
 
 
+async def _handshake(url: str, headers: dict | None) -> dict:
+    """``initialize`` + ``notifications/initialized`` → en-têtes de session.
+
+    Obligatoire avant tout ``tools/call`` contre un serveur à session
+    (ObsiGate) ; sans effet de bord contre un serveur sans état. Les
+    en-têtes retournés portent le ``Mcp-Session-Id`` capturé pendant
+    ``initialize`` quand le serveur en a émis un.
+    """
+    await _rpc(url, {
+        "jsonrpc": "2.0", "id": 1, "method": "initialize",
+        "params": {"protocolVersion": PROTOCOL_VERSION, "capabilities": {},
+                   "clientInfo": CLIENT_INFO},
+    }, headers)
+    session = _session_headers(url, headers)
+    await _notify(url, {"jsonrpc": "2.0", "method": "notifications/initialized"},
+                  session)
+    return session
+
+
 def _server(connector_id: int) -> dict:
     row = get(connector_id)
     if row is None:
         raise ValueError("Serveur MCP introuvable")
-    if row["kind"] not in ("mcp", "tiktak"):
+    if row["kind"] not in MCP_KINDS:
         raise ValueError("Ce connecteur n'est pas un serveur MCP")
     if not row["enabled"]:
         raise ValueError("Serveur MCP désactivé")
@@ -102,6 +214,7 @@ def _normalize(server: dict, raw: list) -> list[dict]:
             "original": str(t["name"]),
             "description": str(t.get("description") or "Outil MCP"),
             "parameters": params,
+            "write": classer_outil(str(t["name"]), t.get("annotations")),
         })
     return out
 
@@ -117,15 +230,9 @@ 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"])
+    session = await _handshake(url, srv["headers"])
     data = await _rpc(url, {"jsonrpc": "2.0", "id": 2, "method": "tools/list",
-                            "params": {}}, srv["headers"])
+                            "params": {}}, session)
     result = data.get("result") if isinstance(data.get("result"), dict) else {}
     tools = _normalize(srv, result.get("tools") or [])
     _save_tools(connector_id, tools)
@@ -137,7 +244,7 @@ def cached_tools() -> list[dict]:
     with get_conn() as conn:
         rows = conn.execute(
             "SELECT id, tools_json FROM agent_connectors "
-            "WHERE kind IN ('mcp', 'tiktak') AND enabled=1"
+            "WHERE kind IN ('mcp', 'tiktak', 'obsigate') AND enabled=1"
         ).fetchall()
     out: list[dict] = []
     for r in rows:
@@ -170,13 +277,18 @@ def _content_text(result: dict) -> str:
 
 
 async def call_tool(connector_id: int, tool: str, arguments: dict | None = None) -> dict:
-    """``tools/call`` → {status, text} (isError → error, borné à 20 000 car.)."""
+    """``tools/call`` → {status, text} (isError → error, borné à 20 000 car.).
+
+    Le handshake (avec session éventuelle) précède systématiquement
+    l'appel : c'est la condition T0 des serveurs à session.
+    """
     srv = _server(connector_id)
     url = _with_secret(srv["url"], connector_id)
+    session = await _handshake(url, srv["headers"])
     data = await _rpc(url, {
         "jsonrpc": "2.0", "id": 3, "method": "tools/call",
         "params": {"name": tool, "arguments": arguments or {}},
-    }, srv["headers"])
+    }, session)
     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}

app/services/permission_manager.py

@@ -55,6 +55,28 @@ DESTRUCTIVE_TOOLS = {
 }
 
 
+def is_write_tool(tool: str) -> bool:
+    """L'outil est-il écrivant ? Outils internes (``WRITE_TOOLS``) et
+    outils **MCP dynamiques** classés écrivants à la découverte
+    (``mcp_client.classer_outil`` — garde-fou n° 2 du pont ObsiGate,
+    ACC-2.2 : avant, un ``mcp_*`` passait toujours pour une lecture,
+    donc un ``apply_*`` distant échappait aux deux portes).
+    """
+    if tool in WRITE_TOOLS:
+        return True
+    if tool.startswith("mcp_"):
+        try:
+            from app.services import mcp_client
+
+            for entry in mcp_client.cached_tools():
+                if entry.get("name") == tool:
+                    return bool(entry.get("write"))
+        except Exception:  # noqa: BLE001 — la classification ne casse rien
+            logger.debug("is_write_tool: classification MCP indisponible",
+                         exc_info=True)
+    return False
+
+
 class PermissionManager:
     """Resolves workspace role and gates agent + granular ACL checks."""
 
@@ -186,7 +208,7 @@ class PermissionManager:
         - destructive tools: admin/owner, or requires confirm approval mode.
         """
         role = self.role_in_workspace(workspace_id)
-        if tool in WRITE_TOOLS and role not in WRITE_ROLES:
+        if is_write_tool(tool) and role not in WRITE_ROLES:
             raise HTTPException(
                 status_code=403,
                 detail=f"Agent tool '{tool}' requires editor+ role (user is '{role}')",

app/services/tool_registry.py

@@ -1230,9 +1230,23 @@ class McpTool(Tool):
     async def execute(self, args, *, user_id=None) -> ToolResult:
         from app.services import mcp_client
 
+        # Garde-fou du pont (ACC-2.2, patron §7.5) : un outil distant
+        # « apply_* » EXÉCUTE une écriture croisée — il n'est jamais
+        # utilisable par un simple appel d'outil depuis un agent
+        # Flowdeck. L'écriture passe par « propose_* » (classé
+        # écrivant : rôles + approbation) puis par l'approbation dans
+        # l'application propriétaire de la donnée.
+        original = self.entry.get("original") or ""
+        if original.lower().startswith("apply_"):
+            return ToolResult(
+                status="error", tool=self.name, target_type="mcp",
+                message=(f"Outil « {original} » bloqué côté client : une "
+                         "écriture croisée ne s'exécute pas par un appel "
+                         "d'outil — propose_* puis approbation dans "
+                         "l'application propriétaire (patron §7.5)."))
         try:
             res = await mcp_client.call_tool(self.entry["connector_id"],
-                                             self.entry.get("original") or "", args)
+                                             original, args)
         except ValueError as exc:
             return ToolResult(status="error", tool=self.name, message=str(exc))
         except Exception as exc:  # noqa: BLE001 — réseau : erreur outil, pas run