from __future__ import annotations import os from typing import Any import gradio as gr import httpx # ── Configuration ───────────────────────────────────────────────────────────── BASE_URL = os.environ.get( "RECHERCHE_DATA_GOUV_API_URL", "https://entrepot.recherche.data.gouv.fr/api", ).rstrip("/") USER_AGENT = "smartbiblia-recherche-data-gouv-demo/0.1" SEARCH_TYPES = ("dataset", "dataverse", "file") METRIC_CATEGORIES = ( "dataverses", "datasets", "files", "downloads", "filedownloads", "uniquedownloads", "uniquefiledownloads", "tree", ) # `pastDays` and `toMonth` need a companion `value` in the path; the demo keeps # the breakdowns that stand on their own. METRIC_BREAKDOWNS = ("monthly", "byCategory", "bySubject", "byType") # A Space has no command line: connector policy is constant here. REQUEST_TIMEOUT = 20.0 # Clamped harder than the canonical server: this endpoint is public. MAX_RESULTS = 10 # One module-level pooled client for the process. HTTP = httpx.Client( timeout=REQUEST_TIMEOUT, follow_redirects=True, headers={"Accept": "application/json", "User-Agent": USER_AGENT}, ) def _get(path: str, params: list[tuple[str, str]] | None = None) -> tuple[Any, str | None]: """GET returning (payload, error). Never raises — the demo answers with data.""" url = f"{BASE_URL}/{path.lstrip('/')}" try: resp = HTTP.get(url, params=params or []) resp.raise_for_status() return resp.json(), None except httpx.HTTPStatusError as exc: return None, f"Recherche Data Gouv returned HTTP {exc.response.status_code}" except httpx.TimeoutException: return None, f"Recherche Data Gouv timed out after {REQUEST_TIMEOUT:g}s" except Exception as exc: # noqa: BLE001 - never crash the Space return None, f"cannot reach Recherche Data Gouv: {exc}" def _normalize_search_item(item: dict) -> dict: """Map one Dataverse search item onto the record shape the server returns.""" return { "source": "recherche-data-gouv", "id": item.get("global_id") or item.get("identifier") or item.get("entity_id"), "type": item.get("type"), "title": item.get("name"), "name": item.get("name"), "description": item.get("description"), "authors": item.get("authors") or [], "subjects": item.get("subjects") or [], "url": item.get("url"), "global_id": item.get("global_id"), "identifier": item.get("identifier"), "published_at": item.get("published_at"), "publisher": item.get("publisher"), "citation": item.get("citation"), "dataverse_alias": item.get("identifier_of_dataverse"), "dataverse_name": item.get("name_of_dataverse"), "file_count": item.get("fileCount"), "version_state": item.get("versionState"), } # ── MCP tools (the only functions exposed with gr.api) ──────────────────────── def search(q: str = "*", type: str | None = None, per_page: int = 5) -> dict: """ Search public Recherche Data Gouv (Dataverse) records: datasets, dataverses and files. Args: q: Solr query, e.g. "biodiversité" or "authorName:Dupont". "*" matches everything. type: Restrict to one entity type — dataset, dataverse or file. Empty for all. per_page: Number of records to return, 1-10 on this demo endpoint. Returns: {"source": "recherche-data-gouv", "command": "search", "query_used": str, "total_found": int, "returned": int, "results": [{"source": str, "id": str, "type": str, "title": str, "authors": [str], "url": str, "published_at": str | null}], "error": str | null} """ out: dict = { "source": "recherche-data-gouv", "command": "search", "query_used": q or "*", "total_found": 0, "returned": 0, "results": [], "error": None, } params: list[tuple[str, str]] = [ ("q", (q or "*").strip() or "*"), ("per_page", str(max(1, min(int(per_page or 5), MAX_RESULTS)))), ("start", "0"), ] if type: if type not in SEARCH_TYPES: out["error"] = "type must be one of " + ", ".join(SEARCH_TYPES) return out params.append(("type", type)) data, error = _get("search", params) if error: out["error"] = error return out payload = data.get("data", {}) if isinstance(data, dict) else {} items = [i for i in (payload.get("items") or []) if isinstance(i, dict)] out["query_used"] = payload.get("q", out["query_used"]) out["total_found"] = payload.get("total_count", 0) out["returned"] = len(items) out["results"] = [_normalize_search_item(i) for i in items] return out def metrics(category: str = "downloads", breakdown: str | None = None) -> dict: """ Fetch a public Dataverse Metrics API counter for the whole Recherche Data Gouv instance. Args: category: Counter to read — dataverses, datasets, files, downloads, filedownloads, uniquedownloads, uniquefiledownloads or tree. breakdown: Optional breakdown of the counter — monthly, byCategory, bySubject or byType. Empty for the total. Returns: {"source": "recherche-data-gouv", "command": "metrics", "category": str, "breakdown": str | null, "data": object, "error": str | null} """ out: dict = { "source": "recherche-data-gouv", "command": "metrics", "category": category, "breakdown": breakdown or None, "data": None, "error": None, } if category not in METRIC_CATEGORIES: out["error"] = "category must be one of " + ", ".join(METRIC_CATEGORIES) return out path = f"info/metrics/{category}" if breakdown: if breakdown not in METRIC_BREAKDOWNS: out["error"] = "breakdown must be one of " + ", ".join(METRIC_BREAKDOWNS) return out path = f"{path}/{breakdown}" data, error = _get(path) if error: out["error"] = error return out out["data"] = data return out # ── Presentation ────────────────────────────────────────────────────────────── def _render_search(payload: dict) -> str: results = payload.get("results") or [] if not results: return "_Aucun enregistrement ne correspond._" lines = [ f"**{payload.get('returned', len(results))} sur {payload.get('total_found', '?')} enregistrements**", "", "| Type | Titre | Auteurs | Publié le |", "|---|---|---|---|", ] for r in results: names = [a for a in (r.get("authors") or []) if a] authors = ", ".join(names[:3]) or "—" if len(names) > 3: authors += " et al." title = (r.get("title") or "Sans titre").replace("|", "\\|") url = r.get("url") lines.append( "| {t} | {title} | {authors} | {pub} |".format( t=r.get("type") or "—", title=f"[{title}]({url})" if url else title, authors=authors.replace("|", "\\|"), pub=(r.get("published_at") or "—")[:10], ) ) return "\n".join(lines) def _render_metrics(payload: dict) -> str: data = payload.get("data") label = payload.get("category") if payload.get("breakdown"): label = f"{label} / {payload['breakdown']}" inner = data.get("data") if isinstance(data, dict) else None # A total comes back as {"status": "OK", "data": {"count": N}}; a breakdown # as a list of buckets. if isinstance(inner, dict) and isinstance(inner.get("count"), (int, float)): return f"**{label}** — `{inner['count']}`" if isinstance(inner, list) and inner: keys = [k for k in inner[0] if isinstance(inner[0], dict)] lines = ["| " + " | ".join(keys) + " |", "|" + "---|" * len(keys)] for row in inner[:25]: lines.append("| " + " | ".join(str(row.get(k, "—")) for k in keys) + " |") return f"**{label}**\n\n" + "\n".join(lines) return f"**{label}** — voir la sortie brute ci-dessous." def _run_search(q, entity_type, per_page): payload = search(q, entity_type or None, per_page) if payload.get("error"): raise gr.Error(payload["error"]) return _render_search(payload), payload def _run_metrics(category, breakdown): payload = metrics(category, breakdown or None) if payload.get("error"): raise gr.Error(payload["error"]) return _render_metrics(payload), payload # ── UI ──────────────────────────────────────────────────────────────────────── with gr.Blocks(title="Recherche Data Gouv MCP demo") as demo: gr.Markdown( "# Recherche Data Gouv MCP demo\n" "Démo autonome du serveur MCP " "[`recherche-data-gouv`](https://github.com/smartbiblia-solutions/agentic-stack/tree/main/mcp/recherche-data-gouv) " ", l'entrepôt Dataverse de la recherche française." ) with gr.Tab("Recherche"): q = gr.Textbox(label="Requête (syntaxe Solr)", value="*", placeholder="biodiversité") with gr.Row(): entity_type = gr.Dropdown( [""] + list(SEARCH_TYPES), value="dataset", label="Type d'entité" ) per_page = gr.Slider(1, MAX_RESULTS, value=5, step=1, label="Résultats") search_btn = gr.Button("Rechercher", variant="primary") search_out = gr.Markdown() search_raw = gr.JSON(label="Sortie brute de l'outil") gr.Examples( examples=[ ["biodiversité", "dataset", 5], ["qzxwvsansresultat", "dataset", 0], ], inputs=[q, entity_type, per_page], label="Une requête qui trouve, une qui ne trouve rien", ) search_btn.click( _run_search, inputs=[q, entity_type, per_page], outputs=[search_out, search_raw], api_name=False, ) with gr.Tab("Métriques"): category = gr.Dropdown( list(METRIC_CATEGORIES), value="datasets", label="Compteur" ) breakdown = gr.Dropdown( [""] + list(METRIC_BREAKDOWNS), value="", label="Ventilation (optionnelle)" ) metrics_btn = gr.Button("Relever", variant="primary") metrics_out = gr.Markdown() metrics_raw = gr.JSON(label="Sortie brute de l'outil") gr.Examples( examples=[["datasets", ""], ["datasets", "bySubject"]], inputs=[category, breakdown], label="Un total, et une ventilation", ) metrics_btn.click( _run_metrics, inputs=[category, breakdown], outputs=[metrics_out, metrics_raw], api_name=False, ) # The only declared MCP tools. Names match the canonical server's. gr.api(search, api_name="search") gr.api(metrics, api_name="metrics") demo.launch(theme=gr.themes.Soft(), mcp_server=True)