| |
|
|
| 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 fetch_trials_v2(term: str, limit: int = 10) -> List[Dict]: |
| """Fetch clinical trials using v2 API.""" |
| params = { |
| "query": term, |
| "pageSize": limit, |
| "fields": "nctId,briefTitle,phase,status,startDate,conditions,interventions" |
| } |
| async with httpx.AsyncClient(timeout=20, headers=HEADERS) as client: |
| r = await client.get(BASE_V2, params=params) |
| if r.status_code == 403: |
| return [] |
| r.raise_for_status() |
| js = r.json() |
| |
| return js.get("studies", []) |
|
|
| 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,Condition,InterventionName,Phase,OverallStatus,StartDate", |
| max_rnk=limit, |
| min_rnk=1, |
| fmt="json", |
| ) |
| 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() |
| js = r.json() |
| return js.get("StudyFieldsResponse", {}).get("StudyFields", []) |
|
|
| 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: |
| trials = await fetch_trials_v1(term, max_studies) |
| return trials |
| except Exception: |
| return [] |
|
|
