mgbam commited on
Commit
19e03c6
·
verified ·
1 Parent(s): 0f5d296

Update mcp/cbio.py

Browse files
Files changed (1) hide show
  1. mcp/cbio.py +15 -54
mcp/cbio.py CHANGED
@@ -1,63 +1,24 @@
1
- # mcp/cbio.py – lightweight async wrapper for cBioPortal REST v3
2
-
3
  """
4
- Fetch mutation summaries for a gene from a chosen study / molecular profile
5
- using the public cBioPortal instance (or a private instance via token).
6
-
7
- * Default public base URL: https://www.cbioportal.org/api
8
- * Optionally read `CBIO_BASE` and `CBIO_KEY` from environment variables to
9
- switch instance and/or add an `Authorization: Bearer <TOKEN>` header.
10
- * Returns the JSON response directly (empty list on any 4xx/5xx).
11
- * Response caching: `@lru_cache` to avoid repeated round‑trips for the same
12
- (gene, profile) pair during a session.
13
  """
14
-
15
- from __future__ import annotations
16
-
17
- import os
18
- import httpx
19
  from functools import lru_cache
20
- from typing import List, Dict
21
-
22
- __all__ = ["fetch_cbio"]
23
 
24
- # ---------------------------------------------------------------------------
25
- # Configuration overridable via env‑vars to support private portals
26
- # ---------------------------------------------------------------------------
27
- _BASE = os.getenv("CBIO_BASE", "https://www.cbioportal.org/api")
28
- _TOKEN = os.getenv("CBIO_KEY")
29
 
30
- _HEADERS: Dict[str, str] = {
31
- "Accept": "application/json",
32
- }
33
- if _TOKEN:
34
- _HEADERS["Authorization"] = f"Bearer {_TOKEN}"
35
-
36
- _URL_TEMPLATE = (
37
- "{base}/molecular-profiles/{profile}/genes/{gene}/mutations?projection=SUMMARY"
38
- )
39
-
40
- # ---------------------------------------------------------------------------
41
- # Public helper
42
- # ---------------------------------------------------------------------------
43
  @lru_cache(maxsize=256)
44
  async def fetch_cbio(
45
  gene: str,
46
  profile: str = "brca_tcga_pan_can_atlas_2018_mutations",
47
- ) -> List[Dict]:
48
- """Return mutation summary list or an empty list on error/empty response."""
49
-
50
- url = _URL_TEMPLATE.format(base=_BASE.rstrip("/"), profile=profile, gene=gene)
51
-
52
- async with httpx.AsyncClient(timeout=15, headers=_HEADERS) as client:
53
- try:
54
- resp = await client.get(url)
55
- # Treat 404/403 as empty but propagate other errors for logging
56
- if resp.status_code == 200:
57
- return resp.json() # type: ignore[return-value]
58
- elif resp.status_code in (404, 403):
59
- return []
60
- resp.raise_for_status()
61
- except Exception:
62
- # network issue, invalid JSON, etc. – keep UX smooth
63
- return []
 
 
 
1
  """
2
+ cbio.py cBioPortal REST helper (no API-key required for public portal).
3
+ Returns variant summary rows for a gene / profile.
 
 
 
 
 
 
 
4
  """
 
 
 
 
 
5
  from functools import lru_cache
6
+ import httpx, asyncio
 
 
7
 
8
+ _BASE = "https://www.cbioportal.org/api/molecular-profiles/{profile}/genes/{gene}/mutations"
9
+ _HEADERS = {"Accept": "application/json"}
 
 
 
10
 
 
 
 
 
 
 
 
 
 
 
 
 
 
11
  @lru_cache(maxsize=256)
12
  async def fetch_cbio(
13
  gene: str,
14
  profile: str = "brca_tcga_pan_can_atlas_2018_mutations",
15
+ ) -> list[dict]:
16
+ if not gene:
17
+ return []
18
+ url = _BASE.format(profile=profile, gene=gene)
19
+ params = {"projection": "SUMMARY"}
20
+ async with httpx.AsyncClient(timeout=10, headers=_HEADERS) as c:
21
+ r = await c.get(url, params=params)
22
+ if r.status_code >= 400:
23
+ return [] # swallow 4xx/5xx empty list
24
+ return r.json()