mcp/clinicaltrials.py

#!/usr/bin/env python3
"""
Fetch clinical trials from ClinicalTrials.gov without external clients.
"""

import httpx
import urllib.parse

async def fetch_clinical_trials(
    query: str,
    max_studies: int = 10
) -> list[dict]:
    """
    Query ClinicalTrials.gov study_fields API for NCTId, Title, Phase, Status.
    Returns up to `max_studies` results.
    """
    # Build the API URL
    base = "https://clinicaltrials.gov/api/query/study_fields"
    params = {
        "expr": query,
        "fields": ",".join(["NCTId", "BriefTitle", "Phase", "OverallStatus"]),
        "max_rnk": max_studies,
        "fmt": "json",
    }
    url = f"{base}?{urllib.parse.urlencode(params)}"

    try:
        async with httpx.AsyncClient(timeout=10) as client:
            resp = await client.get(url)
            resp.raise_for_status()
            data = resp.json()
    except Exception:
        # In case of any network / parse error
        return []

    # Navigate JSON response
    # API docs: https://clinicaltrials.gov/api/gui/ref/study_fields
    studies = data.get("StudyFieldsResponse", {}).get("StudyFields", [])
    output = []
    for s in studies:
        output.append({
            "nctId": s.get("NCTId", [""])[0],
            "briefTitle": s.get("BriefTitle", [""])[0],
            "phase": s.get("Phase", [""])[0],
            "status": s.get("OverallStatus", [""])[0],
        })
    return output