Retrieve all data associated with a UniProtKB entry by accession ID.
Server: https://scp.intern-ai.org.cn/api/v1/mcp/10/Origene-UniProt
Args:
accession (str, required): UniProtKB accession ID (e.g., "P04637" for TP53)
Returns:
Complete protein entry including: protein names, gene names, organism,
sequence, function annotations, subcellular location, post-translational
modifications, disease associations, cross-references.
Retrieve functional enrichment (GO, KEGG, Pfam, InterPro) for a protein set.
Server: https://scp.intern-ai.org.cn/api/v1/mcp/6/Origene-STRING
Args:
identifiers (array of str, required): Gene/protein identifiers (e.g., ["TP53", "MDM2"])
species (int, required): NCBI taxonomy ID (e.g., 9606 for human)
background_string_identifiers (str, required): Background protein set for
enrichment statistics. Use empty string "" for whole-genome background.
Returns:
List of enriched terms with: term name, category (GO/KEGG/Pfam/etc.),
p-value, FDR, description, and which input proteins match.
Retrieve KEGG database entries in flat file format.
Server: https://scp.intern-ai.org.cn/api/v1/mcp/5/Origene-KEGG
Args:
dbentries (str, required): KEGG entry identifier(s). Examples:
- "hsa04115" (p53 signaling pathway)
- "hsa:7157" (TP53 gene entry)
- Multiple entries separated by "+"
option (str, required): Output format. Use "" for default flat file, or:
"aaseq" (amino acid), "ntseq" (nucleotide), "mol", "kcf",
"image", "kgml" (XML), "json"
Returns:
KEGG entry data in the specified format. Flat file includes:
pathway name, description, gene members, compounds, references.
import asyncio
import json
from datetime import datetime
from mcp import ClientSession
from mcp.client.streamable_http import streamablehttp_client
API_KEY = "<YOUR_SCP_HUB_API_KEY>"
SERVERS = {
"tcga": "https://scp.intern-ai.org.cn/api/v1/mcp/11/Origene-TCGA",
"uniprot": "https://scp.intern-ai.org.cn/api/v1/mcp/10/Origene-UniProt",
"string": "https://scp.intern-ai.org.cn/api/v1/mcp/6/Origene-STRING",
"kegg": "https://scp.intern-ai.org.cn/api/v1/mcp/5/Origene-KEGG",
}
GENE = "TP53"
ACCESSION = "P04637"
PATHWAY = "hsa04115"
class OrigeneClient:
def __init__(self, server_url: str, api_key: str):
self.server_url = server_url
self.api_key = api_key
self.session = None
async def connect(self):
try:
self.transport = streamablehttp_client(
url=self.server_url,
headers={"SCP-HUB-API-KEY": self.api_key}
)
self.read, self.write, _ = await self.transport.__aenter__()
self.session_ctx = ClientSession(self.read, self.write)
self.session = await self.session_ctx.__aenter__()
await self.session.initialize()
return True
except Exception as e:
print(f"[ERROR] Connection failed for {self.server_url}: {e}")
return False
async def disconnect(self):
try:
if self.session:
await self.session_ctx.__aexit__(None, None, None)
if hasattr(self, 'transport'):
await self.transport.__aexit__(None, None, None)
except Exception:
pass
def parse_result(self, result):
"""Parse MCP tool result into Python object."""
if isinstance(result, dict):
content_list = result.get("content") or []
else:
content_list = getattr(result, "content", []) or []
texts = []
for item in content_list:
if isinstance(item, dict):
if item.get("type") == "text":
texts.append(item.get("text") or "")
else:
if getattr(item, "type", None) == "text":
texts.append(getattr(item, "text", "") or "")
raw = "".join(texts)
try:
return json.loads(raw)
except (json.JSONDecodeError, TypeError):
return raw
async def main():
report = {
"query": {"gene": GENE, "accession": ACCESSION, "pathway": PATHWAY},
"timestamp": datetime.now().isoformat(),
"steps": {}
}
print(f"[Step 1] Querying TCGA for {GENE} expression across cancers...")
tcga = OrigeneClient(SERVERS["tcga"], API_KEY)
if not await tcga.connect():
report["steps"]["transcriptomics"] = {"error": "Connection failed"}
else:
result = await tcga.session.call_tool(
"get_gene_expression_across_cancers",
arguments={"gene": GENE}
)
data = tcga.parse_result(result)
report["steps"]["transcriptomics"] = data
await tcga.disconnect()
if data and not isinstance(data, str):
print(f" [OK] Received expression data for {GENE}")
else:
print(f" [WARN] Unexpected response format: {str(data)[:200]}")
print(f"[Step 2] Querying UniProt for accession {ACCESSION}...")
uniprot = OrigeneClient(SERVERS["uniprot"], API_KEY)
if not await uniprot.connect():
report["steps"]["proteomics"] = {"error": "Connection failed"}
else:
result = await uniprot.session.call_tool(
"get_uniprotkb_entry_by_accession",
arguments={"accession": ACCESSION}
)
data = uniprot.parse_result(result)
report["steps"]["proteomics"] = data
await uniprot.disconnect()
if data and not isinstance(data, str):
print(f" [OK] Retrieved protein entry for {ACCESSION}")
else:
print(f" [WARN] Unexpected response format: {str(data)[:200]}")
print(f"[Step 3] Running STRING functional enrichment for [{GENE}]...")
string = OrigeneClient(SERVERS["string"], API_KEY)
if not await string.connect():
report["steps"]["enrichment"] = {"error": "Connection failed"}
else:
result = await string.session.call_tool(
"get_functional_enrichment",
arguments={
"identifiers": [GENE],
"species": 9606,
"background_string_identifiers": ""
}
)
data = string.parse_result(result)
report["steps"]["enrichment"] = data
await string.disconnect()
if isinstance(data, list) and len(data) > 0:
print(f" [OK] Found {len(data)} enriched terms")
elif isinstance(data, dict) and data:
print(f" [OK] Enrichment data received")
else:
print(f" [WARN] Unexpected response: {str(data)[:200]}")
print(f"[Step 4] Retrieving KEGG pathway {PATHWAY}...")
kegg = OrigeneClient(SERVERS["kegg"], API_KEY)
if not await kegg.connect():
report["steps"]["kegg_pathway"] = {"error": "Connection failed"}
else:
result = await kegg.session.call_tool(
"kegg_get",
arguments={"dbentries": PATHWAY, "option": ""}
)
data = kegg.parse_result(result)
report["steps"]["kegg_pathway"] = data
await kegg.disconnect()
if data and "p53" in str(data).lower():
print(f" [OK] Retrieved p53 signaling pathway")
elif data:
print(f" [OK] Retrieved pathway data")
else:
print(f" [WARN] Empty response for {PATHWAY}")
output_file = f"{GENE}_multiomics_report.json"
with open(output_file, "w", encoding="utf-8") as f:
json.dump(report, f, indent=2, ensure_ascii=False, default=str)
print(f"\n{'='*60}")
print(f"Multi-omics integration complete!")
print(f"Report saved: {output_file}")
print(f"{'='*60}")
steps = report["steps"]
success = all([
steps.get("transcriptomics") and "error" not in str(steps.get("transcriptomics", "")),
steps.get("proteomics") and "error" not in str(steps.get("proteomics", "")),
steps.get("enrichment") and "error" not in str(steps.get("enrichment", "")),
steps.get("kegg_pathway") and "error" not in str(steps.get("kegg_pathway", "")),
])
if success:
print("[PASS] All 4 omics layers successfully integrated.")
else:
print("[FAIL] Some steps encountered errors. Check the report.")
return success
if __name__ == "__main__":
asyncio.run(main())
