Update mcp/clinicaltrials.py

mcp/clinicaltrials.py

@@ -1,86 +1,31 @@
 """
-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.
+clinicaltrials.py – resilient mirror of ClinicalTrials.gov v2 API.
 """
-
-from __future__ import annotations
-import asyncio, httpx
-from functools import lru_cache
+import httpx, asyncio
 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()
+_BASE = "https://clinicaltrials.gov/api/v2/studies"
+_UA   = "Mozilla/5.0 (MedGenesis; +https://huggingface.co/spaces/mgbam/MCP_Res)"
+_HDRS = {"User-Agent": _UA, "Accept": "application/json"}
+
+async def _fetch(p: Dict, *, retries: int = 2) -> Dict:
+    async with httpx.AsyncClient(timeout=20, headers=_HDRS, follow_redirects=True) as c:
+        for _ in range(retries + 1):
+            r = await c.get(_BASE, params=p)
+            if r.status_code == 403:
+                await asyncio.sleep(1)
+                continue
+            if r.status_code >= 400:
+                return {}
+            return r.json()
+    return {}
 
-# --- 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)
+    p = {
+        "query"   : term,
+        "pageSize": max_studies,
+        "fields"  : ",".join(["nctId","briefTitle","phase","status",
+                              "startDate","conditions","interventions"]),
+    }
+    data = await _fetch(p)
+    return data.get("studies", [])