Create trials.py

genesis/trials.py

@@ -0,0 +1,60 @@
+# genesis/trials.py
+import requests
+from typing import List, Dict
+
+CLINICAL_TRIALS_API = "https://clinicaltrials.gov/api/query/study_fields"
+
+def search_clinical_trials(query: str, max_results: int = 5) -> List[Dict]:
+    """
+    Search ClinicalTrials.gov for trials matching the query.
+    
+    Returns a list of dictionaries with:
+    - title
+    - status
+    - nct_id
+    - locations
+    - url
+    """
+    try:
+        params = {
+            "expr": query,
+            "fields": "NCTId,BriefTitle,OverallStatus,LocationCity,LocationCountry",
+            "min_rnk": 1,
+            "max_rnk": max_results,
+            "fmt": "json"
+        }
+        r = requests.get(CLINICAL_TRIALS_API, params=params, timeout=15)
+        r.raise_for_status()
+        data = r.json()
+
+        studies = data.get("StudyFieldsResponse", {}).get("StudyFields", [])
+        results = []
+        for s in studies:
+            nct_id = s.get("NCTId", [""])[0]
+            title = s.get("BriefTitle", [""])[0]
+            status = s.get("OverallStatus", [""])[0]
+            cities = s.get("LocationCity", [])
+            countries = s.get("LocationCountry", [])
+            locations = [f"{c}, {countries[i]}" if i < len(countries) else c 
+                         for i, c in enumerate(cities)]
+            results.append({
+                "title": title,
+                "status": status,
+                "nct_id": nct_id,
+                "locations": locations,
+                "url": f"https://clinicaltrials.gov/study/{nct_id}"
+            })
+        return results
+    except Exception as e:
+        print(f"[ClinicalTrials] Failed: {e}")
+        return []
+
+def format_trials_markdown(trials: List[Dict]) -> str:
+    """Convert trials list into markdown table."""
+    if not trials:
+        return "No clinical trials found."
+    md = "| Title | Status | NCT ID | Locations |\n|-------|--------|--------|-----------|\n"
+    for t in trials:
+        loc_str = ", ".join(t["locations"]) if t["locations"] else "N/A"
+        md += f"| [{t['title']}]({t['url']}) | {t['status']} | [{t['nct_id']}]({t['url']}) | {loc_str} |\n"
+    return md