#!/usr/bin/env python3
"""Serveur MCP d'Avalyz : lancer un test de recette web depuis Claude (Desktop, Cowork, Code) ou tout client MCP.

Protocole : MCP sur stdio, JSON-RPC 2.0, un message JSON par ligne. Python 3.10 ou plus, bibliothèque standard seulement.
Trois outils :
  avalyz_test_url(url)    lance un test en lecture seule de l'adresse et attend la fin (180 s au plus) ; rend le verdict,
                          les constats et le lien du rapport. Avec AVALYZ_API_KEY : test de ton compte (POST /api/v1/tests).
                          Sans clé : essai public sans inscription (POST /api/essai, 3 par jour).
  avalyz_get_report(id)   relit l'état et le verdict d'un test (compte, ou essai public pour un identifiant d'essai). Un test terminé
                          rend aussi, dans le champ « markdown », le rapport prêt à coller dans une IA pour qu'elle corrige l'app.
  avalyz_list_tests()     tes derniers tests (clé d'API requise).
Variables d'environnement : AVALYZ_API_KEY (clé d'API, jamais écrite dans un dépôt), AVALYZ_BASE (défaut https://avalyz.com).

Installation (1) : télécharge le fichier une fois
    curl -O https://avalyz.com/av/integrations/avalyz_mcp.py

Installation (2a) : Claude Desktop ou Cowork. Ajoute ce bloc au fichier de configuration MCP (« mcpServers »), puis relance l'application :
    {
      "mcpServers": {
        "avalyz": {
          "command": "python3",
          "args": ["/CHEMIN/VERS/avalyz_mcp.py"],
          "env": { "AVALYZ_API_KEY": "TA_CLE" }
        }
      }
    }

Installation (2b) : Claude Code
    claude mcp add avalyz -e AVALYZ_API_KEY=TA_CLE -- python3 /CHEMIN/VERS/avalyz_mcp.py

Sans clé, supprime la ligne « env » (ou « -e … ») : l'outil avalyz_test_url utilise alors l'essai public.
La clé se crée dans Avalyz, page « Clés d'API ». Un test lancé par l'API est en lecture seule : il ne soumet aucun formulaire et ne modifie rien sur le site testé.
"""
import json
import os
import re
import sys
import time
import urllib.error
import urllib.parse
import urllib.request

if sys.version_info < (3, 10):  # pragma: no cover
    sys.stderr.write("avalyz_mcp : Python 3.10 ou plus est requis.\n")
    sys.exit(1)

VERSION = "1.0.0"
PROTOCOLES = ("2025-06-18", "2025-03-26", "2024-11-05")
ATTENTE_MAX = 180      # secondes : durée d'attente d'un test dans avalyz_test_url
PAUSE = 3.0            # secondes entre deux lectures de l'état d'un test
DELAI_HTTP = 30        # secondes : délai d'une requête (jamais une requête longue : on interroge par petits coups)
RX_ID = re.compile(r"[A-Za-z0-9_\-]{8,64}")

OUTILS = [
    {"name": "avalyz_test_url",
     "description": ("Lance un test de recette web EN LECTURE SEULE sur une adresse publique (aucun formulaire soumis, 4 à 6 pages) avec Avalyz, "
                     "attend la fin (180 s au plus) et rend le verdict (GO, GO SOUS CONDITIONS, NO-GO, INCONCLUSIF), les constats par gravité "
                     "(S1 critique à S4 remarque) et le lien du rapport. À appeler après un déploiement."),
     "inputSchema": {"type": "object", "properties": {"url": {"type": "string", "description": "Adresse publique de l'application (https://…)."}},
                     "required": ["url"], "additionalProperties": False},
     "annotations": {"title": "Tester une adresse avec Avalyz", "readOnlyHint": False, "destructiveHint": False, "idempotentHint": False, "openWorldHint": True}},
    {"name": "avalyz_get_report",
     "description": "Relit l'état, le verdict, les constats et le lien du rapport d'un test Avalyz déjà lancé (identifiant rendu par avalyz_test_url ou avalyz_list_tests). Un test terminé rend aussi le champ « markdown » : le rapport structuré pour corriger l'app (issues par gravité, correctifs suggérés, ce qui n'a pas été vérifié).",
     "inputSchema": {"type": "object", "properties": {"id": {"type": "string", "description": "Identifiant du test."}},
                     "required": ["id"], "additionalProperties": False},
     "annotations": {"title": "Lire un rapport Avalyz", "readOnlyHint": True, "openWorldHint": True}},
    {"name": "avalyz_list_tests",
     "description": "Liste les derniers tests du compte Avalyz (identifiant, adresse, état, verdict). Demande AVALYZ_API_KEY.",
     "inputSchema": {"type": "object", "properties": {}, "additionalProperties": False},
     "annotations": {"title": "Lister mes tests Avalyz", "readOnlyHint": True, "openWorldHint": True}},
]


class ErreurOutil(Exception):
    """Une erreur à dire à l'utilisateur (résultat d'outil avec isError, pas une erreur de protocole)."""


def base():
    b = (os.environ.get("AVALYZ_BASE") or "https://avalyz.com").strip().rstrip("/")
    if not re.fullmatch(r"https?://[^\s/?#]+(/[^\s?#]*)?", b):
        raise ErreurOutil("AVALYZ_BASE doit être une adresse http(s) (ex. https://avalyz.com).")
    return b


def cle():
    return (os.environ.get("AVALYZ_API_KEY") or "").strip()


def appeler(methode, chemin, corps=None, avec_cle=False):
    """(code HTTP, JSON décodé ou {}). Lève ErreurOutil si le serveur est injoignable ou répond autre chose que du JSON."""
    h = {"Accept": "application/json", "User-Agent": "avalyz-mcp/" + VERSION}
    donnees = None
    if corps is not None:
        donnees = json.dumps(corps).encode("utf-8")
        h["Content-Type"] = "application/json"
    if avec_cle and cle():
        h["Authorization"] = "Bearer " + cle()
    req = urllib.request.Request(base() + chemin, data=donnees, method=methode, headers=h)
    try:
        with urllib.request.urlopen(req, timeout=DELAI_HTTP) as r:
            code, brut = r.status, r.read(2_000_000)
    except urllib.error.HTTPError as e:
        code, brut = e.code, e.read(2_000_000)
    except (urllib.error.URLError, OSError, ValueError) as e:
        raise ErreurOutil("Avalyz est injoignable (" + base() + ") : " + str(getattr(e, "reason", e))[:200]) from None
    try:
        d = json.loads(brut.decode("utf-8") or "{}")
    except ValueError:
        raise ErreurOutil(f"Réponse inattendue d'Avalyz (HTTP {code}).") from None
    return code, d if isinstance(d, dict) else {}


def _message(code, d):
    m = str(d.get("error") or "")[:300]
    if code == 401:
        return "Clé d'API refusée ou absente : vérifie AVALYZ_API_KEY (page « Clés d'API » d'Avalyz)."
    if code == 429 and not m:
        return "Quota atteint ou trop de tests en cours : réessaie plus tard."
    if code == 404 and not m:
        return "Introuvable (service éteint ou identifiant inconnu)."
    return m or f"Avalyz a répondu HTTP {code}."


def normaliser(d, ident):
    """Même forme pour les deux sources (API de compte, essai public) : id, etat, url, verdict, pourquoi, constats, rapport."""
    out = {"id": d.get("id") or ident, "etat": d.get("etat") or "inconnu", "url": d.get("url") or ""}
    for k in ("verdict", "pourquoi"):
        if d.get(k):
            out[k] = d[k]
    if isinstance(d.get("constats"), list):
        out["constats"] = [{"severite": str(c.get("severite", "")), "titre": str(c.get("titre", ""))[:300]} for c in d["constats"] if isinstance(c, dict)][:50]
    if d.get("erreur"):
        out["erreur"] = str(d["erreur"])[:300]
    out["rapport"] = base() + (d.get("rapport_url") or "/r/" + str(out["id"]))
    if out["etat"] == "en_cours":
        out["note"] = "Le test tourne encore : relis-le avec avalyz_get_report."
    return out


def _markdown(ident, compte):
    """Le rapport au format « Copier pour mon IA » (Markdown), ou None si le serveur ne le propose pas (version plus ancienne)."""
    try:
        if compte:
            code, d = appeler("GET", "/api/v1/tests/" + urllib.parse.quote(ident, safe="") + "/report.md", avec_cle=True)
        else:
            code, d = appeler("GET", "/api/essai/" + urllib.parse.quote(ident, safe="") + "/ia")
    except ErreurOutil:
        return None
    md = d.get("markdown") if code == 200 else None
    return md if isinstance(md, str) and md else None


def _avec_markdown(res, ident, compte):
    if res.get("etat") == "termine":
        md = _markdown(ident, compte)
        if md:
            res["markdown"] = md
    return res


def _etat(ident, compte, avec_md=False):
    if compte:
        code, d = appeler("GET", "/api/v1/tests/" + urllib.parse.quote(ident, safe=""), avec_cle=True)
        if code == 200:
            res = normaliser(d, ident)
            return _avec_markdown(res, ident, True) if avec_md else res
        if code != 404:
            raise ErreurOutil(_message(code, d))
    code, d = appeler("GET", "/api/essai/" + urllib.parse.quote(ident, safe=""))
    if code == 200:
        d = dict(d, id=ident, rapport_url="/r/" + ident)
        res = normaliser(d, ident)
        return _avec_markdown(res, ident, False) if avec_md else res
    raise ErreurOutil("Test inconnu." if code == 404 else _message(code, d))


def _url_valide(url):
    if not isinstance(url, str) or not url.strip():
        raise ErreurOutil("Paramètre « url » requis : l'adresse publique de l'application (https://…).")
    url = url.strip()
    if len(url) > 300:
        raise ErreurOutil("Adresse trop longue (300 caractères au plus).")
    return url


def outil_test_url(args):
    url = _url_valide(args.get("url"))
    fin = time.monotonic() + ATTENTE_MAX
    compte = bool(cle())
    if compte:
        code, d = appeler("POST", "/api/v1/tests", {"url": url, "attendre": False}, avec_cle=True)
        if code != 200 or not d.get("id"):
            raise ErreurOutil(_message(code, d))
        ident = d["id"]
        res = normaliser(d, ident)
    else:
        code, d = appeler("POST", "/api/essai", {"url": url})
        if code != 200 or not d.get("jeton"):
            raise ErreurOutil(_message(code, d))
        ident = d["jeton"]
        res = {"id": ident, "etat": "en_cours", "url": url, "rapport": base() + "/r/" + ident}
    while res.get("etat") == "en_cours" and time.monotonic() < fin:
        time.sleep(min(PAUSE, max(0.0, fin - time.monotonic())))
        try:
            res = _etat(ident, compte)
        except ErreurOutil:
            if time.monotonic() >= fin:
                break
    if res.get("etat") == "en_cours":
        res["note"] = f"Pas fini après {ATTENTE_MAX} s : relis-le avec avalyz_get_report (id {ident})."
    return res


def outil_get_report(args):
    ident = args.get("id")
    if not isinstance(ident, str) or not RX_ID.fullmatch(ident.strip()):
        raise ErreurOutil("Paramètre « id » invalide : indique l'identifiant rendu par avalyz_test_url.")
    return _etat(ident.strip(), bool(cle()), avec_md=True)


def outil_list_tests(args):
    if not cle():
        raise ErreurOutil("avalyz_list_tests demande AVALYZ_API_KEY (les essais publics n'ont pas de liste).")
    code, d = appeler("GET", "/api/v1/tests/", avec_cle=True)
    if code == 404:
        raise ErreurOutil("Ce serveur Avalyz ne propose pas encore la liste des tests : utilise avalyz_get_report avec un identifiant.")
    if code != 200:
        raise ErreurOutil(_message(code, d))
    tests = [normaliser(t, t.get("id", "")) for t in d.get("tests", []) if isinstance(t, dict)][:50]
    return {"tests": tests}


FONCTIONS = {"avalyz_test_url": outil_test_url, "avalyz_get_report": outil_get_report, "avalyz_list_tests": outil_list_tests}


def _ok(i, resultat):
    return {"jsonrpc": "2.0", "id": i, "result": resultat}


def _err(i, code, message):
    return {"jsonrpc": "2.0", "id": i, "error": {"code": code, "message": message}}


def traiter(msg):
    """Un message JSON-RPC -> la réponse (dict), ou None pour une notification."""
    if not isinstance(msg, dict) or msg.get("jsonrpc") != "2.0" or not isinstance(msg.get("method"), str):
        return _err(msg.get("id") if isinstance(msg, dict) else None, -32600, "Requête invalide.")
    if "id" not in msg:
        return None  # notification (notifications/initialized, notifications/cancelled…) : ignorée
    i, m, p = msg["id"], msg["method"], msg.get("params")
    p = p if isinstance(p, dict) else {}
    if m == "initialize":
        v = p.get("protocolVersion")
        return _ok(i, {"protocolVersion": v if v in PROTOCOLES else PROTOCOLES[0], "capabilities": {"tools": {"listChanged": False}},
                       "serverInfo": {"name": "avalyz", "title": "Avalyz", "version": VERSION},
                       "instructions": "Teste une application web avec Avalyz (lecture seule) : avalyz_test_url après chaque déploiement, puis lis le verdict et les constats."})
    if m == "ping":
        return _ok(i, {})
    if m == "tools/list":
        return _ok(i, {"tools": OUTILS})
    if m == "tools/call":
        f = FONCTIONS.get(p.get("name"))
        if not f:
            return _err(i, -32602, "Outil inconnu : " + str(p.get("name"))[:80])
        a = p.get("arguments")
        if a is not None and not isinstance(a, dict):
            return _err(i, -32602, "« arguments » doit être un objet.")
        try:
            r = f(a or {})
            return _ok(i, {"content": [{"type": "text", "text": json.dumps(r, ensure_ascii=False, indent=2)}], "isError": False})
        except ErreurOutil as e:
            return _ok(i, {"content": [{"type": "text", "text": str(e)}], "isError": True})
        except Exception as e:  # jamais de trace ni de clé dans la réponse
            return _ok(i, {"content": [{"type": "text", "text": "Erreur inattendue (" + type(e).__name__ + ")."}], "isError": True})
    return _err(i, -32601, "Méthode inconnue : " + m[:80])


def boucle(entree=None, sortie=None):
    entree, sortie = entree or sys.stdin, sortie or sys.stdout
    for ligne in entree:
        ligne = ligne.strip()
        if not ligne:
            continue
        try:
            msg = json.loads(ligne)
        except ValueError:
            rep = _err(None, -32700, "JSON invalide.")
        else:
            try:
                rep = [r for r in (traiter(x) for x in msg) if r] if isinstance(msg, list) and msg else traiter(msg)
            except Exception as e:  # pragma: no cover
                rep = _err(msg.get("id") if isinstance(msg, dict) else None, -32603, "Erreur interne (" + type(e).__name__ + ").")
        if rep:
            sortie.write(json.dumps(rep, ensure_ascii=False) + "\n")
            sortie.flush()


if __name__ == "__main__":
    try:
        sys.stdin.reconfigure(encoding="utf-8")
        sys.stdout.reconfigure(encoding="utf-8")
    except (AttributeError, ValueError):
        pass
    try:
        boucle()
    except KeyboardInterrupt:
        pass