Update mcp/clinicaltrials.py

mcp/clinicaltrials.py

@@ -1,56 +1,86 @@
-#!/usr/bin/env python3
-"""MedGenesis – ClinicalTrials.gov **v2** async wrapper.
-
-The legacy v1 JSON endpoint often throttles (HTTP 403/503).  This helper
-switches to the production **v2** REST API, adds retry/back‑off, and
-mirrors the interface expected by existing code (`search_trials`).
-
-Key points
-~~~~~~~~~~
-* Endpoint: `https://clinicaltrials.gov/api/v2/studies` (public, no key).
-* Back‑off retry (2×, 4×) for 429/5xx.
-* Results cached 12 h (512 queries).
-* Returns list of dicts with minimal fields: `nctId`, `briefTitle`,
-  `phase`, `status`, `startDate`, `conditions`, `interventions`.
 """
-from __future__ import annotations
+clinicaltrials.py  ·  Modernised helper (July-2025-ready)
+
+Order of endpoints
+──────────────────
+1. beta-ut (modern ingest)  https://beta-ut.clinicaltrials.gov/api/v2/studies
+2. prod v2                 https://clinicaltrials.gov/api/v2/studies
+3. legacy v1               https://clinicaltrials.gov/api/query/study_fields
+4. WHO ICTRP mirror        https://trialsearch.who.int/api/StudyFields   (JSON)
+
+All calls are GET, JSON; no API-key required.
+
+Returns [] on any failure so orchestrator never raises.
+"""
 
+from __future__ import annotations
 import asyncio, httpx
 from functools import lru_cache
-from typing import List, Dict, Any
-
-_BASE = "https://clinicaltrials.gov/api/v2/studies"
-_TIMEOUT = 15
-_HEADERS = {"User-Agent": "MedGenesis/1.0 (https://huggingface.co/spaces)"}
-_MAX_PAGE = 50  # absolute hard cap
-
-# ---------------------------------------------------------------------
-# Internal fetch with retry
-# ---------------------------------------------------------------------
-async def _fetch(params: Dict[str, Any], *, retries: int = 3) -> List[Dict]:
-    delay = 2
-    for _ in range(retries):
-        async with httpx.AsyncClient(timeout=_TIMEOUT, headers=_HEADERS) as cli:
-            r = await cli.get(_BASE, params=params)
-            if r.status_code == 200:
-                return r.json().get("studies", [])
-            if r.status_code in {429, 500, 503}:
-                await asyncio.sleep(delay)
-                delay *= 2
-                continue
-            r.raise_for_status()
-    return []  # final fallback
-
-# ---------------------------------------------------------------------
-# Public helper – cached
-# ---------------------------------------------------------------------
-@lru_cache(maxsize=512)
-async def search_trials(term: str, *, max_studies: int = 10) -> List[Dict]:
-    """Return up to *max_studies* clinical‑trial dicts for free‑text *term*."""
-    max_studies = max(1, min(max_studies, _MAX_PAGE))
-    params = {
-        "query"   : term,
-        "pageSize": max_studies,
-        "fields"  : "nctId,briefTitle,phase,status,startDate,conditions,interventions",
-    }
-    return await _fetch(params)
+from typing import List, Dict
+
+_UA = ("Mozilla/5.0 (Windows NT 10.0; Win64; x64) "
+       "AppleWebKit/537.36 (KHTML, like Gecko) "
+       "Chrome/126.0 Safari/537.36")
+
+_HDR = {"User-Agent": _UA, "Accept": "application/json"}
+_TIMEOUT = 12
+_RETRY   = 1
+_BETA    = "https://beta-ut.clinicaltrials.gov/api/v2/studies"
+_V2      = "https://clinicaltrials.gov/api/v2/studies"
+_V1      = "https://clinicaltrials.gov/api/query/study_fields"
+_WHO     = "https://trialsearch.who.int/api/StudyFields"
+
+# ────────────────────────────────────────────────────────────────────
+async def _get(url: str, params: Dict) -> Dict:
+    async with httpx.AsyncClient(timeout=_TIMEOUT,
+                                 headers=_HDR,
+                                 follow_redirects=True) as cli:
+        r = await cli.get(url, params=params)
+        r.raise_for_status()
+        return r.json()
+
+# --- individual endpoint helpers -----------------------------------
+async def _try_beta(term: str, n: int) -> List[Dict]:
+    p = {"query": term, "pageSize": n,
+         "fields": "nctId,briefTitle,phase,status,startDate,conditions,interventions"}
+    return (await _get(_BETA, p)).get("studies", [])
+
+async def _try_v2(term: str, n: int) -> List[Dict]:
+    p = {"query": term, "pageSize": n,
+         "fields": "nctId,briefTitle,phase,status,startDate,conditions,interventions"}
+    return (await _get(_V2, p)).get("studies", [])
+
+async def _try_v1(term: str, n: int) -> List[Dict]:
+    p = {"expr": term,
+         "fields": ("NCTId,BriefTitle,Phase,OverallStatus,StartDate,"
+                    "Condition,InterventionName"),
+         "min_rnk": 1, "max_rnk": n, "fmt": "json"}
+    j = (await _get(_V1, p)).get("StudyFieldsResponse", {})
+    return j.get("StudyFields", [])
+
+async def _try_who(term: str, n: int) -> List[Dict]:
+    p = {"expr": term, "fields": "URL,HealthCondition,PublicTitle",
+         "min_rnk": 1, "max_rnk": n, "fmt": "json"}
+    j = (await _get(_WHO, p))
+    return j.get("StudyFieldsResponse", {}).get("StudyFields", [])
+
+# ────────────────────────────────────────────────────────────────────
+@lru_cache(maxsize=256)
+async def search_trials(term: str, *, max_studies: int = 20) -> List[Dict]:
+    """
+    Return ≤max_studies trial records using BETA→V2→V1→WHO cascade.
+    Empty list on total failure.
+    """
+    limit = max(1, min(max_studies, 100))
+    for fn in (_try_beta, _try_v2, _try_v1, _try_who):
+        for attempt in range(_RETRY + 1):
+            try:
+                return await fn(term, limit)
+            except (httpx.HTTPStatusError, httpx.ReadTimeout):
+                if attempt < _RETRY:
+                    await asyncio.sleep(0.8)
+    return []
+
+# Back-compat alias (some older code calls search_trials_v2)
+async def search_trials_v2(term: str, *, max_studies: int = 20):
+    return await search_trials(term, max_studies=max_studies)