Update mcp/ctgov.py

mcp/ctgov.py

@@ -1,21 +1,71 @@
-"""ClinicalTrials.gov v2 async helper.
-   Docs: https://clinicaltrials.gov/api/gui/ref/api_urls#v2
-   Provides `search_trials_v2` which returns a list of study dicts.
+#!/usr/bin/env python3
 """
+ClinicalTrials.gov helper – v2 primary, v1 fallback.
+"""
+
+from __future__ import annotations
+import asyncio, httpx, urllib.parse
+from functools import lru_cache
+from typing import List, Dict, Any
 
-import httpx, asyncio
-from typing import List, Dict
+# v2 REST
+_V2 = "https://clinicaltrials.gov/api/v2/studies"
+# legacy v1 (read-only but still works)
+_V1 = "https://clinicaltrials.gov/api/query/study_fields"
 
-_BASE = "https://clinicaltrials.gov/api/v2/studies"
+_HEADERS = {
+    "User-Agent": (
+        "Mozilla/5.0 (Windows NT 10.0; Win64; x64) "
+        "AppleWebKit/537.36 (KHTML, like Gecko) "
+        "Chrome/125.0 Safari/537.36"
+    ),
+    "Accept": "application/json",
+}
+_TIMEOUT = 15
+_MAX = 50  # hard cap
 
-async def search_trials_v2(query: str, *, max_n: int = 20) -> List[Dict]:
-    """Search CT.gov v2 endpoint and return list of studies."""
+async def _get(url: str, params: Dict[str, Any]) -> httpx.Response:
+    async with httpx.AsyncClient(timeout=_TIMEOUT, headers=_HEADERS) as cli:
+        return await cli.get(url, params=params)
+
+async def _try_v2(term: str, limit: int) -> List[Dict]:
     params = {
-        "query": query,
-        "pageSize": max_n,
-        "fields": "nctId,briefTitle,phase,status,startDate,conditions,interventions",
+        "query"   : term,
+        "pageSize": limit,
+        "fields"  : "nctId,briefTitle,phase,status,startDate,conditions,interventions",
     }
-    async with httpx.AsyncClient(timeout=15) as client:
-        resp = await client.get(_BASE, params=params)
-        resp.raise_for_status()
-        return resp.json().get("studies", [])
+    r = await _get(_V2, params)
+    if r.status_code == 200:
+        return r.json().get("studies", [])
+    if r.status_code in {403, 429, 500, 503}:
+        raise RuntimeError("v2 unavailable")
+    r.raise_for_status()       # other 4xx
+
+async def _try_v1(term: str, limit: int) -> List[Dict]:
+    params = {
+        "expr"    : term,
+        "fields"  : "NCTId,BriefTitle,Phase,OverallStatus,StartDate,Condition,InterventionName",
+        "max_rnk" : limit,
+        "min_rnk" : 1,
+        "fmt"     : "json",
+    }
+    r = await _get(_V1, params)
+    r.raise_for_status()
+    return r.json().get("StudyFieldsResponse", {}).get("StudyFields", [])
+
+@lru_cache(maxsize=512)
+async def search_trials(term: str, *, max_studies: int = 10) -> List[Dict]:
+    """
+    Return up to *max_studies* trial dicts; v2 first, v1 fallback.
+    """
+    limit = max(1, min(max_studies, _MAX))
+    # v2 with back-off: 2s → 4s
+    for delay in (0, 2, 4):
+        try:
+            if delay:
+                await asyncio.sleep(delay)
+            return await _try_v2(term, limit)
+        except RuntimeError:
+            continue
+    # fallback to legacy endpoint
+    return await _try_v1(term, limit)