mcp/cbio.py

"""
cBioPortal helper · REST v4 · zero-crash

* Public instance needs **no** API-key  :contentReference[oaicite:0]{index=0}
* Optional Bearer token (`CBIO_KEY`) for private portals (e.g. GENIE) :contentReference[oaicite:1]{index=1}
* One retry on 5xx / ReadTimeout, else returns []
"""

from __future__ import annotations
import os, asyncio, httpx
from functools import lru_cache
from typing import List, Dict

_TOKEN = os.getenv("CBIO_KEY")          # optional
_HDR   = {"Accept": "application/json"}
if _TOKEN:
    _HDR["Authorization"] = f"Bearer {_TOKEN}"

_URL = ("https://www.cbioportal.org/api"
        "/molecular-profiles/{profile}/genes/{gene}/mutations?projection=SUMMARY"
        )                               # swagger ref :contentReference[oaicite:2]{index=2}

@lru_cache(maxsize=256)
async def fetch_cbio(
    gene    : str,
    profile : str = "brca_tcga_pan_can_atlas_2018_mutations"
) -> List[Dict]:
    async def _hit() -> 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:
                return []               # gene absent
            r.raise_for_status()
            return r.json()

    try:
        return await _hit()
    except (httpx.HTTPStatusError, httpx.ReadTimeout):
        await asyncio.sleep(0.5)        # one back-off
        try:
            return await _hit()
        except Exception:
            return []                   # graceful fallback