Update mcp/clinicaltrials.py

mcp/clinicaltrials.py

@@ -1,53 +1,62 @@
-import httpx, random
+# mcp/clinicaltrials.py
 
-_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",
+import httpx
+from typing import List, Dict
+
+BASE_V2 = "https://clinicaltrials.gov/api/v2/studies"
+BASE_V1 = "https://clinicaltrials.gov/api/query/study_fields"
+
+HEADERS = {
+    "User-Agent": (
+        "Mozilla/5.0 (Windows NT 10.0; Win64; x64) "
+        "AppleWebKit/537.36 (KHTML, like Gecko) "
+        "Chrome/124.0 Safari/537.36"
+    )
 }
 
-async def _try_v2(term: str, n: int):
-    p = {
+async def fetch_trials_v2(term: str, limit: int = 10) -> List[Dict]:
+    """Fetch clinical trials using v2 API."""
+    params = {
         "query": term,
-        "pageSize": n,
-        "fields": ",".join([
-            "nctId", "briefTitle", "phase", "status",
-            "startDate", "conditions", "interventions",
-        ]),
+        "pageSize": limit,
+        "fields": "nctId,briefTitle,phase,status,startDate,conditions,interventions"
     }
-    async with httpx.AsyncClient(
-        headers={"User-Agent": _HEADERS[random.randint(0,2)]}, timeout=12
-    ) as c:
-        r = await c.get(_BASE_V2, params=p)
+    async with httpx.AsyncClient(timeout=20, headers=HEADERS) as client:
+        r = await client.get(BASE_V2, params=params)
         if r.status_code == 403:
-            raise RuntimeError("v2 blocked")
+            return []  # API blocked or IP rate-limited
         r.raise_for_status()
-        return r.json().get("studies", [])
+        js = r.json()
+        # List of studies under ['studies']
+        return js.get("studies", [])
 
-async def _try_v1(term: str, n: int):
-    p = dict(
+async def fetch_trials_v1(term: str, limit: int = 10) -> List[Dict]:
+    """Fallback: Fetch trials using legacy v1 API."""
+    params = dict(
         expr=term,
-        fields="NCTId,BriefTitle,Phase,OverallStatus,StartDate,Condition,InterventionName",
-        max_rnk=n, min_rnk=1, fmt="json",
+        fields="NCTId,BriefTitle,Condition,InterventionName,Phase,OverallStatus,StartDate",
+        max_rnk=limit,
+        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)
+    async with httpx.AsyncClient(timeout=20, headers=HEADERS) as client:
+        r = await client.get(BASE_V1, params=params)
+        if r.status_code == 403:
+            return []
         r.raise_for_status()
-        return r.json()["StudyFieldsResponse"]["StudyFields"]
+        js = r.json()
+        return js.get("StudyFieldsResponse", {}).get("StudyFields", [])
 
-# public
-async def search_trials(term: str, max_studies: int = 20):
+async def search_trials(term: str, max_studies: int = 10) -> List[Dict]:
+    """Unified entry point. Try V2, fallback to V1 if needed."""
+    try:
+        trials = await fetch_trials_v2(term, max_studies)
+        if trials:
+            return trials
+    except Exception:
+        pass
     try:
-        return await _try_v2(term, max_studies)
+        trials = await fetch_trials_v1(term, max_studies)
+        return trials
     except Exception:
-        try:
-            return await _try_v1(term, max_studies)
-        except Exception:
-            return []        # always return list
+        return []