Update mcp/clinicaltrials.py

mcp/clinicaltrials.py

@@ -1,31 +1,53 @@
-"""
-clinicaltrials.py – resilient mirror of ClinicalTrials.gov v2 API.
-"""
-import httpx, asyncio
-from typing import List, Dict
+import httpx, random
 
-_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"}
+_BASE_V2 = "https://clinicaltrials.gov/api/v2/studies"
+_BASE_V1 = "https://clinicaltrials.gov/api/query/study_fields"
+_HEADERS = {
+    # 3 random desktop UAs – simple rotation avoids naïve geo blocks
+    0: "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 "
+       "(KHTML, like Gecko) Chrome/125 Safari/537.36",
+    1: "Mozilla/5.0 (Macintosh; Intel Mac OS X 12_6) AppleWebKit/605.1.15 "
+       "(KHTML, like Gecko) Version/16 Safari/605.1.15",
+    2: "Mozilla/5.0 (X11; Linux x86_64) Gecko/20100101 Firefox/126.0",
+}
 
-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 {}
-
-async def search_trials(term: str, *, max_studies: int = 20) -> List[Dict]:
+async def _try_v2(term: str, n: int):
     p = {
-        "query"   : term,
-        "pageSize": max_studies,
-        "fields"  : ",".join(["nctId","briefTitle","phase","status",
-                              "startDate","conditions","interventions"]),
+        "query": term,
+        "pageSize": n,
+        "fields": ",".join([
+            "nctId", "briefTitle", "phase", "status",
+            "startDate", "conditions", "interventions",
+        ]),
     }
-    data = await _fetch(p)
-    return data.get("studies", [])
+    async with httpx.AsyncClient(
+        headers={"User-Agent": _HEADERS[random.randint(0,2)]}, timeout=12
+    ) as c:
+        r = await c.get(_BASE_V2, params=p)
+        if r.status_code == 403:
+            raise RuntimeError("v2 blocked")
+        r.raise_for_status()
+        return r.json().get("studies", [])
+
+async def _try_v1(term: str, n: int):
+    p = dict(
+        expr=term,
+        fields="NCTId,BriefTitle,Phase,OverallStatus,StartDate,Condition,InterventionName",
+        max_rnk=n, min_rnk=1, fmt="json",
+    )
+    async with httpx.AsyncClient(
+        headers={"User-Agent": _HEADERS[random.randint(0,2)]}, timeout=12
+    ) as c:
+        r = await c.get(_BASE_V1, params=p)
+        r.raise_for_status()
+        return r.json()["StudyFieldsResponse"]["StudyFields"]
+
+# public
+async def search_trials(term: str, max_studies: int = 20):
+    try:
+        return await _try_v2(term, max_studies)
+    except Exception:
+        try:
+            return await _try_v1(term, max_studies)
+        except Exception:
+            return []        # always return list