| """ | |
| clinicaltrials.py – resilient mirror of ClinicalTrials.gov v2 API. | |
| """ | |
| import httpx, asyncio | |
| from typing import List, Dict | |
| _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 {} | |
| async def search_trials(term: str, *, max_studies: int = 20) -> List[Dict]: | |
| p = { | |
| "query" : term, | |
| "pageSize": max_studies, | |
| "fields" : ",".join(["nctId","briefTitle","phase","status", | |
| "startDate","conditions","interventions"]), | |
| } | |
| data = await _fetch(p) | |
| return data.get("studies", []) | |