Create cbio.py

mcp/cbio.py

@@ -0,0 +1,41 @@
+"""
+cBioPortal async helper – returns cohort-level mutation summary.
+
+• Works for ANY public molecular-profile ID
+• One retry on 5xx, returns [] on 404/timeout
+• Optional CBIO_KEY (JWT) for private studies
+"""
+
+from functools import lru_cache
+import os, asyncio, httpx
+
+_BASE = "https://www.cbioportal.org/api"
+_URL  = _BASE + (
+    "/molecular-profiles/{profile}/genes/{gene}/mutations?projection=SUMMARY"
+)
+_KEY  = os.getenv("CBIO_KEY")            # optional
+
+@lru_cache(maxsize=256)
+async def fetch_cbio(gene: str,
+                     profile: str = "brca_tcga_pan_can_atlas_2018_mutations"
+                    ) -> list[dict]:
+    hdr = {"Accept": "application/json"}
+    if _KEY:
+        hdr["Authorization"] = f"Bearer {_KEY}"
+
+    async def _once() -> list[dict]:
+        async with httpx.AsyncClient(timeout=10, headers=hdr) as c:
+            r = await c.get(_URL.format(profile=profile, gene=gene))
+            if r.status_code == 404:        # gene not present
+                return []
+            r.raise_for_status()
+            return r.json()
+
+    try:
+        return await _once()
+    except httpx.HTTPStatusError:
+        await asyncio.sleep(0.4)            # back-off then retry
+        try:
+            return await _once()
+        except Exception:                   # still bad → empty payload
+            return []