mcp/cbio.py

"""
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 []