from __future__ import annotations import os from typing import Any, Literal import gradio as gr import httpx # ── Configuration ───────────────────────────────────────────────────────────── API_BASE_URL = os.environ.get("IDREF_API_URL", "https://idref-resolver.smartbiblia.fr/").rstrip("/") # The credential is read from the environment only: never a tool argument, never # echoed into a payload. API_KEY = os.environ.get("IDREF_API_KEY", "") # A Space has no command line: connector policy is constant here. An alignment # fans out to as many as 41 upstream ABES requests, so the budget is generous. REQUEST_TIMEOUT = 180.0 # Clamped harder than the canonical server (which allows 100 / 20): this endpoint # is public and every call spends the operator's API budget. MAX_CANDIDATES = 30 MAX_RETURNED = 10 # Mirrors the API's `EmbeddingModel` Literal. A closed enum there is a closed # enum here, so a wrong value fails in the client instead of costing a round trip. EMBEDDING_MODELS = ( "lexical", "lexical-idf", "albert-bge-m3", "granite", "qwen", "minilm", ) EmbeddingModel = Literal[ "lexical", "lexical-idf", "albert-bge-m3", "granite", "qwen", "minilm" ] # One module-level pooled client for the process. HTTP = httpx.Client(timeout=REQUEST_TIMEOUT, follow_redirects=True) def _post(path: str, payload: dict) -> tuple[dict | None, str | None]: """POST returning (data, error). Never raises — the demo answers with data.""" headers = {"Content-Type": "application/json"} if API_KEY: headers["X-API-Key"] = API_KEY try: resp = HTTP.post(f"{API_BASE_URL}{path}", json=payload, headers=headers) except httpx.TimeoutException: return None, f"the API did not answer within {REQUEST_TIMEOUT:g}s" except httpx.HTTPError as exc: return None, f"cannot reach the API at {API_BASE_URL}: {exc}" if resp.status_code >= 400: # Surface the API's own detail; never echo the key back. return None, f"API {resp.status_code}: {resp.text[:300]}" try: return resp.json(), None except ValueError: return None, f"API returned a non-JSON body: {resp.text[:200]}" # ── MCP tools (the only functions exposed with gr.api) ──────────────────────── def align_person( name: str, works: list[str] | None = None, field: str = "", affiliation: str = "", year: str = "", context: str = "", embedding_model: EmbeddingModel = "lexical-idf", max_candidates: int = 20, accept_threshold: float = 0.65, margin_threshold: float = 0.08, max_returned_candidates: int = 5, ) -> dict: """ Align a person to an IdRef PPN from their name plus any disambiguating context. `best_ppn` is populated only when `status == "accepted"`. On any other status the alignment abstained: report the abstention, do not fall back to the top candidate's PPN. Raising a threshold makes the service stricter, i.e. more abstentions. Args: name: Full person name, e.g. "Valérie Robert". Required. works: Titles of documents the person is linked to. Often decisive. field: Discipline or subject area, e.g. "sociologie des sciences". affiliation: Institution, laboratory or place, e.g. "Université de Nancy". year: A relevant year as a string, e.g. "2003". context: Any other free text — a biographical note, an abstract. embedding_model: How texts are compared — lexical, lexical-idf, albert-bge-m3, granite, qwen or minilm. max_candidates: Candidate authorities to enrich and score, 1-30 on this demo endpoint. accept_threshold: Minimum final score for "accepted" (0.0-1.0). margin_threshold: Minimum lead over the runner-up for "accepted" (0.0-1.0). max_returned_candidates: Ranked candidates to include in the reply, 0-10 on this demo endpoint. Returns: {"source": "idref-resolver-api", "command": "align_person", "status": "accepted" | "ambiguous" | "low_confidence" | "not_found" | null, "best_ppn": str | null, "best_candidate": object | null, "candidates": [object], "candidates_scored": int, "similarity": object | null, "error": str | null} """ reply: dict[str, Any] = { "source": "idref-resolver-api", "command": "align_person", "status": None, "best_ppn": None, "best_candidate": None, "candidates": [], "candidates_scored": 0, "similarity": None, "error": None, } if not (name or "").strip(): reply["error"] = "name is required" return reply if embedding_model not in EMBEDDING_MODELS: reply["error"] = "embedding_model must be one of " + ", ".join(EMBEDDING_MODELS) return reply payload = { "name": name.strip(), "works": [w for w in (works or []) if w], "field": field or "", "affiliation": affiliation or "", "role": "", "year": str(year or ""), "context": context or "", "embedding_model": embedding_model, "max_candidates": max(1, min(int(max_candidates or 20), MAX_CANDIDATES)), "accept_threshold": float(accept_threshold), "margin_threshold": float(margin_threshold), } result, error = _post("/align/person", payload) if error: reply["error"] = error return reply ranked = result.get("candidates") or [] reply.update( status=result.get("status"), best_ppn=result.get("best_ppn"), best_candidate=result.get("best_candidate"), candidates=ranked[: max(0, min(int(max_returned_candidates or 5), MAX_RETURNED))], candidates_scored=len(ranked), similarity=result.get("similarity"), error=result.get("error"), ) return reply # ── Presentation ────────────────────────────────────────────────────────────── VERDICTS = { "accepted": "✅ **Accepté** — un PPN est retenu.", "ambiguous": "🤔 **Ambigu** — plusieurs autorités se tiennent ; l'outil s'abstient.", "low_confidence": "⚠️ **Confiance insuffisante** — le meilleur score reste sous le seuil.", "not_found": "❌ **Aucune autorité candidate** pour ce nom.", } def _render(payload: dict) -> str: lines: list[str] = [] status = payload.get("status") lines.append(VERDICTS.get(status, f"**Statut** — `{status}`")) if payload.get("best_ppn"): ppn = payload["best_ppn"] lines.append(f"\n**PPN retenu** — [{ppn}](https://www.idref.fr/{ppn})") elif status: lines.append( "\n_Abstention : ne retenez pas le premier candidat comme s'il avait " "été accepté. Une abstention est une réponse._" ) if payload.get("error"): lines.append(f"\n> ⚠️ {payload['error']}") candidates = payload.get("candidates") or [] if candidates: lines += [ "", f"**{len(candidates)} candidats affichés sur {payload.get('candidates_scored', '?')} évalués**", "", "| PPN | Score final | Forme retenue |", "|---|---|---|", ] for c in candidates: ppn = c.get("ppn") or "—" score = ((c.get("score") or {}).get("final")) score_txt = f"{score:.3f}" if isinstance(score, (int, float)) else "—" forms = ((c.get("evidence") or {}).get("preferred_forms") or []) form = (forms[0] if forms else "—").replace("|", "\\|") link = f"[{ppn}](https://www.idref.fr/{ppn})" if ppn != "—" else "—" lines.append(f"| {link} | {score_txt} | {form} |") return "\n".join(lines) def _run(name, works_text, field, affiliation, year, context, embedding_model, max_candidates, accept_threshold, margin_threshold): works = [line.strip() for line in (works_text or "").splitlines() if line.strip()] payload = align_person( name=name, works=works, field=field, affiliation=affiliation, year=year, context=context, embedding_model=embedding_model, max_candidates=max_candidates, accept_threshold=accept_threshold, margin_threshold=margin_threshold, ) if payload.get("error") and payload.get("status") is None: raise gr.Error(payload["error"]) return _render(payload), payload # ── UI ──────────────────────────────────────────────────────────────────────── with gr.Blocks(title="IdRef resolver MCP demo") as demo: gr.Markdown( "# IdRef resolver — démo MCP\n" "Démo autonome du serveur MCP " "[`idref-resolver-api`](https://github.com/smartbiblia-solutions/agentic-stack/tree/main/mcp/idref-resolver-api) " ", aligner une autoritéé personne extraite d'un document sur son PPN IdRef, ou " "s'abstenir quand l'indice est trop faible. " ) with gr.Row(): with gr.Column(): name = gr.Textbox(label="Nom de la personne", placeholder="Valérie Robert") works_text = gr.Textbox( label="Titres d'œuvres (un par ligne)", lines=3, placeholder="Nous n'avons jamais été modernes", ) field = gr.Textbox(label="Domaine", placeholder="sociologie des sciences") affiliation = gr.Textbox(label="Affiliation", placeholder="Université de Nancy") with gr.Row(): year = gr.Textbox(label="Année", placeholder="2003") embedding_model = gr.Dropdown( list(EMBEDDING_MODELS), value="lexical-idf", label="Comparaison" ) context = gr.Textbox(label="Contexte libre", lines=2) with gr.Accordion("Seuils", open=False): max_candidates = gr.Slider( 1, MAX_CANDIDATES, value=20, step=1, label="Candidats évalués" ) accept_threshold = gr.Slider( 0.0, 1.0, value=0.65, step=0.01, label="Seuil d'acceptation" ) margin_threshold = gr.Slider( 0.0, 1.0, value=0.08, step=0.01, label="Écart minimal au second" ) align_btn = gr.Button("Aligner", variant="primary") with gr.Column(): verdict_out = gr.Markdown() raw_out = gr.JSON(label="Sortie brute de l'outil") inputs = [name, works_text, field, affiliation, year, context, embedding_model, max_candidates, accept_threshold, margin_threshold] gr.Examples( examples=[ ["Bruno Latour", "Nous n'avons jamais été modernes", "sociologie des sciences", "Sciences Po", "1991", "", "lexical-idf", 20, 0.65, 0.08], ["Michel Foucault", "Surveiller et punir : naissance de la prison", "philosophie", "Collège de France", "1975", "", "lexical-idf", 20, 0.65, 0.08], ["Valérie Robert", "", "études germaniques", "Université Sorbonne Nouvelle", "", "", "lexical-idf", 20, 0.65, 0.08], ["Bruno Latour", "Nous n'avons jamais été modernes", "sociologie des sciences", "Sciences Po", "1991", "", "albert-bge-m3", 20, 0.65, 0.08], ["Bruno Latour", "Nous n'avons jamais été modernes", "sociologie des sciences", "Sciences Po", "1991", "", "lexical-idf", 20, 0.85, 0.20], ], inputs=inputs, label="Titre décisif, autre titre décisif, indices sans œuvre, comparaison " "sémantique, puis le même cas avec des seuils sévères", ) align_btn.click(_run, inputs=inputs, outputs=[verdict_out, raw_out], api_name=False) # The only declared MCP tool. The name matches the canonical server's. gr.api(align_person, api_name="align_person") demo.launch(theme=gr.themes.Soft(), mcp_server=True)