mcp/clinicaltrials.py

# mcp/clinicaltrials.py  –  403-proof CPU-only helper
import httpx, asyncio, datetime
from typing import List, Dict

BASE = "https://clinicaltrials.gov/api/query/study_fields"
UA   = ("Mozilla/5.0 (Windows NT 10.0; Win64; x64) "
        "AppleWebKit/537.36 (KHTML, like Gecko) "
        "Chrome/124.0 Safari/537.36")  # real browser UA

_HEADERS = {"User-Agent": UA}

async def _fetch(url: str, params: Dict) -> Dict:
    async with httpx.AsyncClient(timeout=20, headers=_HEADERS) as c:
        r = await c.get(url, params=params)
        # If still 4xx/5xx, return empty dict – keep app alive
        if r.status_code != 200:
            return {}
        return r.json()

async def search_trials(term: str, max_studies: int = 10) -> List[Dict]:
    params = dict(
        expr=term,
        fields="NCTId,BriefTitle,Condition,InterventionName,Phase,OverallStatus,StartDate",
        max_rnk=max_studies,
        min_rnk=1,
        fmt="json",
    )
    data = await _fetch(BASE, params)
    return data.get("StudyFieldsResponse", {}).get("StudyFields", [])