Update mcp/cbio.py
Browse files- mcp/cbio.py +11 -33
mcp/cbio.py
CHANGED
@@ -1,36 +1,14 @@
|
|
1 |
-
|
|
|
|
|
2 |
|
3 |
-
|
4 |
-
|
5 |
-
|
6 |
-
_CBIO_ENABLED = True
|
7 |
-
except ImportError:
|
8 |
-
logging.warning("pybioportal package not available; disabling cBioPortal integration")
|
9 |
-
_CBIO_ENABLED = False
|
10 |
|
11 |
-
|
12 |
-
""
|
13 |
-
|
14 |
-
|
15 |
-
"""
|
16 |
-
if not _CBIO_ENABLED:
|
17 |
-
return []
|
18 |
|
19 |
-
|
20 |
-
# 1) Grab all molecular profiles (you can restrict to particular studies if you like)
|
21 |
-
profiles = fetch_molecular_profiles() # returns list of dicts with 'molecular_profile_id', etc.
|
22 |
-
|
23 |
-
# 2) For each profile, fetch the mutations for our gene
|
24 |
-
variants = []
|
25 |
-
for prof in profiles:
|
26 |
-
mpid = prof.get("molecular_profile_id")
|
27 |
-
if mpid:
|
28 |
-
# fetch_muts_in_multiple_mol_profs returns a list of mutation dicts
|
29 |
-
muts = fetch_muts_in_multiple_mol_profs([mpid], gene=gene)
|
30 |
-
variants.extend(muts or [])
|
31 |
-
|
32 |
-
return variants
|
33 |
-
|
34 |
-
except Exception as e:
|
35 |
-
logging.warning("cBioPortal variant fetch failed: %s", e)
|
36 |
-
return []
|
|
|
1 |
+
# mcp/cbio.py
|
2 |
+
from functools import lru_cache
|
3 |
+
from mcp.clients import BaseClient
|
4 |
|
5 |
+
class CbioClient(BaseClient):
|
6 |
+
def __init__(self):
|
7 |
+
super().__init__("https://www.cbioportal.org/api", None)
|
|
|
|
|
|
|
|
|
8 |
|
9 |
+
@lru_cache(256)
|
10 |
+
async def fetch_variants(self, gene: str, profile: str="brca_tcga_pan_can_atlas_2018_mutations"):
|
11 |
+
path = f"molecular-profiles/{profile}/genes/{gene}/mutations?projection=SUMMARY"
|
12 |
+
return await self.request("GET", path)
|
|
|
|
|
|
|
13 |
|
14 |
+
cbio = CbioClient()
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|