Update mcp/clinicaltrials.py

mcp/clinicaltrials.py

@@ -1,62 +1,34 @@
 # mcp/clinicaltrials.py
 
-import httpx
-from typing import List, Dict
+from pytrials.client import ClinicalTrials
+import asyncio
 
-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_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
+        )
     )
-}
-
-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 []  # API blocked or IP rate-limited
-        r.raise_for_status()
-        js = r.json()
-        # List of studies under ['studies']
-        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 []
+    # 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
+    ]