#!/usr/bin/env python3 # -*- coding: utf-8 -*- """MCP « Mémoire des crises » de Saper Vedere (stdio, JSON-RPC 2.0, bibliothèque standard seule). Expose la base des crises numériques documentées par ReputatioLab (internal-pages/crises-data/) : les cartes (cards.json), les fiches détaillées (fiches/.json) et les statistiques (stats.json). Mécanique de transport reprise du squelette validé sur ce poste : ~/.claude/skills/ocr/scripts/mcp_stdio.py (en-têtes Content-Length). Échoue FERMÉ au démarrage si les données manquent. """ from __future__ import annotations import ast import difflib import json import os import sys import tempfile import unicodedata from typing import Any PROTOCOL_VERSION = "2025-03-26" HERE = os.path.dirname(os.path.abspath(__file__)) SITE = os.environ.get("MEMOIRE_CRISES_SITE", "https://www.saper-vedere.eu/internal-pages/") # Donnees : le dossier local du depot quand il existe (poste de travail), sinon les MEMES fichiers servis par le site, # rapatries a la demande dans un cache temporaire. C'est ce qui permet de distribuer ce seul fichier au public. DATA = os.environ.get("MEMOIRE_CRISES_DATA") or os.path.normpath(os.path.join(HERE, "..", "..", "internal-pages", "crises-data")) CACHE = os.path.join(tempfile.gettempdir(), "sv-memoire-crises") EN_LIGNE = not os.path.exists(os.path.join(DATA, "cards.json")) def rapatrier(nom: str) -> str | None: """Telecharge crises-data/ depuis le site dans le cache et rend le chemin local ; None si le site ne l'a pas.""" import urllib.error import urllib.request chemin = os.path.join(CACHE, nom.replace("/", os.sep)) if os.path.exists(chemin): return chemin os.makedirs(os.path.dirname(chemin), exist_ok=True) req = urllib.request.Request(SITE + "crises-data/" + nom, headers={"User-Agent": "sv-memoire-crises-mcp/1.1"}) try: with urllib.request.urlopen(req, timeout=60) as resp: corps = resp.read() except urllib.error.HTTPError as exc: if exc.code == 404: return None raise SystemExit(f"memoire-crises : {SITE}crises-data/{nom} -> HTTP {exc.code}") from exc except Exception as exc: # noqa: BLE001 - un MCP sans donnees echoue ferme, avec l'URL raise SystemExit(f"memoire-crises : {SITE}crises-data/{nom} injoignable ({exc})") from exc if not corps.lstrip().startswith((b"{", b"[")): return None # une page HTML (deploiement en cours) n'est pas une donnee with open(chemin, "wb") as fh: fh.write(corps) return chemin def chemin_donnee(nom: str) -> str | None: """Chemin local de crises-data/ : depot, ou cache alimente depuis le site.""" if not EN_LIGNE: p = os.path.join(DATA, nom.replace("/", os.sep)) return p if os.path.exists(p) else None return rapatrier(nom) CHAMPS = ("secteur", "niveau", "typologie", "origine", "valeur", "declencheur", "reseau", "format", "annee", "sphere_libaert") CLE = {"secteur": "sector", "niveau": "level", "typologie": "typologie", "origine": "origine", "valeur": "valeur", "declencheur": "triggerCat", "format": "format", "annee": "year", "sphere_libaert": "sphere_libaert"} def sans_accents(s: Any) -> str: s = "" if s is None else str(s) return "".join(c for c in unicodedata.normalize("NFD", s) if unicodedata.category(c) != "Mn").lower() def charger() -> tuple[list[dict[str, Any]], dict[str, dict[str, Any]], dict[str, Any]]: chemins = {f: chemin_donnee(f) for f in ("cards.json", "stats.json")} for f, p in chemins.items(): if not p: ou = f"{SITE}crises-data/" if EN_LIGNE else DATA raise SystemExit(f"memoire-crises : fichier introuvable {f} (donnees attendues dans {ou})") if not EN_LIGNE and not os.path.isdir(os.path.join(DATA, "fiches")): raise SystemExit(f"memoire-crises : dossier introuvable {os.path.join(DATA, 'fiches')}") with open(chemins["cards.json"], encoding="utf-8") as fh: cartes = json.load(fh) with open(chemins["stats.json"], encoding="utf-8") as fh: stats = json.load(fh) if not isinstance(cartes, list) or not cartes: raise SystemExit("memoire-crises : cards.json vide ou mal forme") index = {c["slug"]: c for c in cartes if c.get("slug")} for c in cartes: c["_txt"] = sans_accents(" ".join(str(c.get(k) or "") for k in ("org", "description", "sector", "trigger", "triggerCat", "valeur", "domain"))) return cartes, index, stats CARTES, INDEX, STATS = charger() def fiche_de(slug: str) -> dict[str, Any] | None: if slug not in INDEX: # jamais de requete au site pour un slug inconnu return None p = chemin_donnee("fiches/" + slug + ".json") if not p: return None with open(p, encoding="utf-8") as fh: f = json.load(fh) # les fiches portent des listes/dicts serialises en texte Python : on les remet en objets for k, v in list(f.items()): if isinstance(v, str) and v[:1] in "[{" and v[-1:] in "]}": try: f[k] = ast.literal_eval(v) except Exception: pass return f def carte_publique(c: dict[str, Any]) -> dict[str, Any]: out = {k: v for k, v in c.items() if not k.startswith("_")} out["url"] = SITE + "memoire-crise-fiche.html?slug=" + c["slug"] return out def filtrer(a: dict[str, Any]) -> list[dict[str, Any]]: res = CARTES texte = sans_accents(a.get("texte") or "") if texte: mots = texte.split() res = [c for c in res if all(m in c["_txt"] for m in mots)] org = sans_accents(a.get("organisation") or "") if org: res = [c for c in res if org in sans_accents(c.get("org"))] if a.get("annee") is not None: res = [c for c in res if c.get("year") == int(a["annee"])] if a.get("annee_min") is not None: res = [c for c in res if (c.get("year") or 0) >= int(a["annee_min"])] if a.get("annee_max") is not None: res = [c for c in res if (c.get("year") or 0) <= int(a["annee_max"])] if a.get("niveau") is not None: res = [c for c in res if int(c.get("level") or 0) == int(a["niveau"])] for champ in ("secteur", "typologie", "origine", "valeur", "declencheur", "format", "sphere_libaert"): val = a.get(champ) if val: v = sans_accents(val) res = [c for c in res if sans_accents(c.get(CLE[champ])) == v] if a.get("reseau"): v = sans_accents(a["reseau"]) res = [c for c in res if any(v in sans_accents(r) for r in (c.get("reseaux") or []))] return sorted(res, key=lambda c: (-(c.get("year") or 0), -(c.get("month") or 0), sans_accents(c.get("org")))) def valeurs(champ: str) -> list[Any]: if champ == "reseau": vals = {r for c in CARTES for r in (c.get("reseaux") or [])} else: vals = {c.get(CLE[champ]) for c in CARTES} vals.discard(None) vals.discard("") return sorted(vals, key=lambda x: (str(type(x)), x)) # ------------------------------------------------------------------ outils def t_rechercher(a: dict[str, Any]) -> dict[str, Any]: res = filtrer(a) limite = max(1, min(int(a.get("limite") or 20), 100)) page = max(1, int(a.get("page") or 1)) debut = (page - 1) * limite return {"total_base": len(CARTES), "total_correspondant": len(res), "page": page, "limite": limite, "pages": (len(res) + limite - 1) // limite, "resultats": [carte_publique(c) for c in res[debut:debut + limite]]} def t_fiche(a: dict[str, Any]) -> dict[str, Any]: slug = (a.get("slug") or "").strip() if not slug: raise ValueError("slug obligatoire (le slug d'une carte rendue par rechercher_crises)") c = INDEX.get(slug) if not c: proches = difflib.get_close_matches(slug, list(INDEX.keys()), n=5, cutoff=0.4) return {"trouve": False, "slug": slug, "message": "aucune crise avec ce slug", "slugs_proches": proches} f = fiche_de(slug) return {"trouve": True, "carte": carte_publique(c), "fiche": f, "fiche_detaillee_presente": f is not None} def t_statistiques(a: dict[str, Any]) -> dict[str, Any]: par = a.get("par") or "annee" if par not in CHAMPS: raise ValueError(f"par doit valoir une de : {', '.join(CHAMPS)}") res = filtrer(a) comptes: dict[str, int] = {} for c in res: cles = c.get("reseaux") or [] if par == "reseau" else [c.get(CLE[par])] for k in cles: k = "" if k is None else str(k) comptes[k] = comptes.get(k, 0) + 1 ordre = sorted(comptes.items(), key=lambda kv: (-kv[1], kv[0])) if par != "annee" else sorted(comptes.items(), key=lambda kv: kv[0]) return {"par": par, "total_base": len(CARTES), "total_correspondant": len(res), "comptes": [{"valeur": k, "nombre": v} for k, v in ordre]} def t_organisation(a: dict[str, Any]) -> dict[str, Any]: nom = sans_accents(a.get("nom") or "") if not nom: raise ValueError("nom obligatoire") res = sorted([c for c in CARTES if nom in sans_accents(c.get("org"))], key=lambda c: ((c.get("year") or 0), (c.get("month") or 0))) orgs = sorted({c.get("org") for c in res}) return {"nom_cherche": a.get("nom"), "organisations_correspondantes": orgs, "nombre_de_crises": len(res), "total_base": len(CARTES), "crises": [carte_publique(c) for c in res]} def t_valeurs(a: dict[str, Any]) -> dict[str, Any]: champ = a.get("champ") if champ not in CHAMPS: raise ValueError(f"champ doit valoir une de : {', '.join(CHAMPS)}") v = valeurs(champ) return {"champ": champ, "nombre": len(v), "valeurs": v} def t_similaires(a: dict[str, Any]) -> dict[str, Any]: slug = (a.get("slug") or "").strip() c = INDEX.get(slug) if not c: raise ValueError("slug inconnu : utilise fiche_crise pour obtenir des slugs proches") limite = max(1, min(int(a.get("limite") or 5), 30)) cand = [x for x in CARTES if x["slug"] != slug and x.get("org") != c.get("org") and x.get("sector") == c.get("sector") and x.get("typologie") == c.get("typologie")] cand.sort(key=lambda x: (x.get("triggerCat") != c.get("triggerCat"), x.get("level") != c.get("level"), abs((x.get("year") or 0) - (c.get("year") or 0)))) return {"reference": carte_publique(c), "critere": "meme secteur et meme typologie, autre organisation ; declencheur, niveau et annee proches d'abord", "total_correspondant": len(cand), "similaires": [carte_publique(x) for x in cand[:limite]]} def _aplatir(x: Any) -> str: """Les fiches melangent textes, listes de textes et listes d'objets ({title, desc}, {date, text}) : tout devient du texte.""" if isinstance(x, list): return "\n".join(t for t in (_aplatir(v) for v in x) if t) if isinstance(x, dict): return " : ".join(t for t in (_aplatir(v) for v in x.values()) if t) return x.strip() if isinstance(x, str) else "" def _titre(c: dict[str, Any]) -> str: return f"{c.get('org')} ({c.get('year') or '?'}) : {c.get('description') or ''}" def t_search(a: dict[str, Any]) -> dict[str, Any]: """Contrat des connecteurs ChatGPT (recherche approfondie) : search(query) -> results[{id, title, url}].""" res = filtrer({"texte": a.get("query") or ""})[:20] return {"results": [{"id": c["slug"], "title": _titre(c)[:200], "url": carte_publique(c)["url"]} for c in res]} def t_fetch(a: dict[str, Any]) -> dict[str, Any]: """Contrat des connecteurs ChatGPT : fetch(id) -> {id, title, text, url, metadata}.""" slug = (a.get("id") or "").strip() c = INDEX.get(slug) if not c: raise ValueError("id inconnu : utilise search pour obtenir des identifiants (slugs)") f = fiche_de(slug) or {} blocs = [("Résumé", c.get("description")), ("Sous-titre", f.get("heroSubtitle")), ("Contexte", f.get("context")), ("Synthèse", f.get("summary")), ("Chronologie", f.get("timeline")), ("Réponse de l'organisation", f.get("response")), ("Conséquences", f.get("consequences")), ("Résolution", f.get("resolution")), ("Leçons", f.get("lessons"))] parties = [t + "\n" + v for t, v in ((t, _aplatir(v)) for t, v in blocs) if v] return {"id": slug, "title": _titre(c), "text": "\n\n".join(parties) or (c.get("description") or ""), "url": carte_publique(c)["url"], "metadata": {"organisation": c.get("org"), "annee": c.get("year"), "secteur": c.get("sector"), "niveau": c.get("level"), "typologie": c.get("typologie"), "origine": c.get("origine"), "declencheur": c.get("triggerCat"), "reseaux": c.get("reseaux")}} OUTILS: dict[str, Any] = { "rechercher_crises": t_rechercher, "fiche_crise": t_fiche, "statistiques": t_statistiques, "organisation": t_organisation, "valeurs_possibles": t_valeurs, "similaires": t_similaires, "search": t_search, "fetch": t_fetch, } FILTRES = { "texte": {"type": "string", "description": "Recherche plein texte (organisation, description, secteur, declencheur), insensible aux accents ; tous les mots doivent etre presents"}, "organisation": {"type": "string", "description": "Nom (ou fragment) de l'organisation"}, "annee": {"type": "integer", "description": "Annee exacte (2004 a 2025)"}, "annee_min": {"type": "integer"}, "annee_max": {"type": "integer"}, "secteur": {"type": "string", "description": "Valeur exacte : voir valeurs_possibles('secteur')"}, "niveau": {"type": "integer", "description": "1, 2 ou 3 (3 = le plus grave)"}, "typologie": {"type": "string", "description": "Communicationnelle, Fonctionnelle, Technique, Juridique, Accidentelle"}, "origine": {"type": "string", "description": "Interne ou Externe"}, "valeur": {"type": "string", "description": "Valeur mobilisee (Critique, Indignation, Feminisme...) : voir valeurs_possibles('valeur')"}, "declencheur": {"type": "string", "description": "Categorie de declencheur : voir valeurs_possibles('declencheur')"}, "reseau": {"type": "string", "description": "Reseau de propagation (fragment accepte : Twitter, Facebook, TikTok...)"}, "format": {"type": "string"}, "sphere_libaert": {"type": "string"}, } def tools() -> list[dict[str, Any]]: return [ {"name": "rechercher_crises", "description": f"Cherche dans les {len(CARTES)} crises numeriques documentees (2004-2025) avec des filtres cumulatifs. Rend des CARTES (resume court) triees par date decroissante, avec le denominateur total_correspondant / total_base. Ne rend pas la fiche detaillee : appeler fiche_crise(slug).", "inputSchema": {"type": "object", "properties": {**FILTRES, "limite": {"type": "integer", "description": "1 a 100, defaut 20"}, "page": {"type": "integer", "description": "defaut 1"}}}}, {"name": "fiche_crise", "description": "La fiche complete d'une crise par son slug : contexte, chronologie, parties prenantes, reponse de l'organisation, consequences, lecons, crises similaires, URL publique. Si le slug n'existe pas, rend trouve=false et cinq slugs proches (ne pas inventer).", "inputSchema": {"type": "object", "properties": {"slug": {"type": "string"}}, "required": ["slug"]}}, {"name": "statistiques", "description": "Comptes par annee, secteur, niveau, typologie, origine, valeur, declencheur, reseau, format ou sphere_libaert, sur la base entiere ou filtree (memes filtres que rechercher_crises). Toujours avec le total.", "inputSchema": {"type": "object", "properties": {"par": {"type": "string", "enum": list(CHAMPS)}, **FILTRES}}}, {"name": "organisation", "description": "Toutes les crises d'une organisation (recidives), par date croissante, avec le compte. Le nom est cherche en fragment insensible aux accents.", "inputSchema": {"type": "object", "properties": {"nom": {"type": "string"}}, "required": ["nom"]}}, {"name": "valeurs_possibles", "description": "Les valeurs reellement presentes dans la base pour un champ de filtre (secteur, niveau, typologie, origine, valeur, declencheur, reseau, format, annee, sphere_libaert). A appeler avant de filtrer plutot que de deviner une valeur.", "inputSchema": {"type": "object", "properties": {"champ": {"type": "string", "enum": list(CHAMPS)}}, "required": ["champ"]}}, {"name": "similaires", "description": "Crises comparables a une crise donnee : meme secteur et meme typologie, autres organisations, declencheur et niveau proches d'abord.", "inputSchema": {"type": "object", "properties": {"slug": {"type": "string"}, "limite": {"type": "integer", "description": "1 a 30, defaut 5"}}, "required": ["slug"]}}, {"name": "search", "description": "Search the crisis database (full text, accent-insensitive). Returns up to 20 results with id (slug), title and url. Standard connector tool for ChatGPT deep research; same data as rechercher_crises.", "inputSchema": {"type": "object", "properties": {"query": {"type": "string", "description": "Search query"}}, "required": ["query"]}}, {"name": "fetch", "description": "Fetch the full document of one crisis by id (slug from search): text (context, response, consequences, lessons, timeline), url and metadata. Standard connector tool for ChatGPT deep research.", "inputSchema": {"type": "object", "properties": {"id": {"type": "string", "description": "Crisis id (slug)"}}, "required": ["id"]}}, ] def mcp_text(payload: Any, is_error: bool = False) -> dict[str, Any]: text = payload if isinstance(payload, str) else json.dumps(payload, ensure_ascii=False, indent=1) return {"content": [{"type": "text", "text": text}], "isError": is_error} def handle(message: dict[str, Any]) -> dict[str, Any] | None: method = message.get("method") msg_id = message.get("id") params = message.get("params") or {} if method == "initialize": return {"jsonrpc": "2.0", "id": msg_id, "result": {"protocolVersion": PROTOCOL_VERSION, "capabilities": {"tools": {"listChanged": False}}, "serverInfo": {"name": "sv-memoire-crises", "version": "1.0.0"}}} if method in ("notifications/initialized", "initialized"): return None if method == "ping": return {"jsonrpc": "2.0", "id": msg_id, "result": {}} if method == "tools/list": return {"jsonrpc": "2.0", "id": msg_id, "result": {"tools": tools()}} if method == "tools/call": name = params.get("name") args = params.get("arguments") or {} fn = OUTILS.get(name) if not fn: return {"jsonrpc": "2.0", "id": msg_id, "result": mcp_text(f"outil inconnu : {name}. Outils : {', '.join(OUTILS)}", True)} try: return {"jsonrpc": "2.0", "id": msg_id, "result": mcp_text(fn(args))} except (ValueError, TypeError) as exc: return {"jsonrpc": "2.0", "id": msg_id, "result": mcp_text(f"parametre invalide : {exc}", True)} except Exception as exc: # noqa: BLE001 - un MCP echoue ferme return {"jsonrpc": "2.0", "id": msg_id, "result": mcp_text(f"erreur interne : {exc}", True)} if msg_id is None: return None return {"jsonrpc": "2.0", "id": msg_id, "error": {"code": -32601, "message": f"methode inconnue : {method}"}} MODE = {"ndjson": True} # transport reconnu au premier message : les clients MCP (Claude, Cursor) parlent un JSON par ligne def read_message() -> dict[str, Any] | None: """Un message par ligne (JSON-RPC newline-delimited, transport stdio du MCP) ; l'ancien cadrage Content-Length (style LSP) reste accepte, et la reponse est ecrite dans le meme cadrage que la question. Le 17/09/2026, ce serveur ne parlait que Content-Length : Claude Code attendait une ligne JSON et le serveur attendait une ligne vide, chacun sans fin (« connection timed out after 30000ms »).""" headers: dict[str, str] = {} while True: line = sys.stdin.buffer.readline() if not line: return None if line.lstrip().startswith(b"{"): MODE["ndjson"] = True return json.loads(line.decode("utf-8")) if line in (b"\r\n", b"\n"): if headers: break continue decoded = line.decode("utf-8", "replace") if ":" not in decoded: continue key, value = decoded.split(":", 1) headers[key.strip().lower()] = value.strip() MODE["ndjson"] = False length = int(headers.get("content-length") or "0") if length <= 0: return None raw = sys.stdin.buffer.read(length) return json.loads(raw.decode("utf-8")) def write_message(message: dict[str, Any]) -> None: raw = json.dumps(message, ensure_ascii=False).encode("utf-8") if MODE["ndjson"]: sys.stdout.buffer.write(raw + b"\n") else: sys.stdout.buffer.write(f"Content-Length: {len(raw)}\r\n\r\n".encode("ascii") + raw) sys.stdout.buffer.flush() def main() -> None: while True: try: message = read_message() except Exception as exc: # noqa: BLE001 write_message({"jsonrpc": "2.0", "id": None, "error": {"code": -32700, "message": str(exc)}}) continue if message is None: return reply = handle(message) if reply is not None: write_message(reply) if __name__ == "__main__": main()