mcp/clinicaltrials.py

# mcp/clinicaltrials.py

from pytrials.client import ClinicalTrials
import asyncio

async def fetch_clinical_trials(query: str, max_studies: int = 10) -> list[dict]:
    """
    Pulls NCTId, Title, Phase, and Status for studies matching the given query.
    Uses the pytrials wrapper over ClinicalTrials.gov.    
    """
    # pytrials expects a direct search expression, e.g. disease MeSH term or drug
    client = ClinicalTrials()
    # These are the fields you want from the API 
    study_fields = ["NCTId", "BriefTitle", "Phase", "OverallStatus"]
    # Wrap in a ThreadPool so we can call it in asyncio
    loop = asyncio.get_event_loop()
    records = await loop.run_in_executor(
        None,
        lambda: client.get_study_fields(
            search_expr=query,
            fields=study_fields,
            max_studies=max_studies
        )
    )
    # Normalize the field names for your UI
    return [
        {
            "nctId": rec.get("NCTId"),
            "briefTitle": rec.get("BriefTitle"),
            "phase": rec.get("Phase"),
            "status": rec.get("OverallStatus"),
        }
        for rec in records
    ]