microbiome-genomics

name	microbiome_genomics
description	Microbiome Genomics Analysis - Analyze microbial genome: NCBI genome data, taxonomy, KEGG metabolic pathways, and annotation. Use this skill for metagenomics tasks involving get genome dataset report by taxon get taxonomy kegg find get genome annotation report. Combines 4 tools from 2 SCP server(s).
i18n	{"zh":{"description":"微生物基因组分析：NCBI数据。"}}

Microbiome Genomics Analysis

Discipline: Metagenomics | Tools Used: 4 | Servers: 2

Description

Analyze microbial genome: NCBI genome data, taxonomy, KEGG metabolic pathways, and annotation.

Tools Used

• get_genome_dataset_report_by_taxon from ncbi-server (streamable-http) - https://scp.intern-ai.org.cn/api/v1/mcp/9/Origene-NCBI
• get_taxonomy from ncbi-server (streamable-http) - https://scp.intern-ai.org.cn/api/v1/mcp/9/Origene-NCBI
• kegg_find from kegg-server (streamable-http) - https://scp.intern-ai.org.cn/api/v1/mcp/5/Origene-KEGG
• get_genome_annotation_report from ncbi-server (streamable-http) - https://scp.intern-ai.org.cn/api/v1/mcp/9/Origene-NCBI

Workflow

1. Get genome dataset for E. coli
2. Get taxonomic classification
3. Find KEGG metabolic pathways
4. Get genome annotation

Test Case

Input

{
    "taxon": "Escherichia coli",
    "accession": "GCF_000005845.2"
}

Expected Steps

1. Get genome dataset for E. coli
2. Get taxonomic classification
3. Find KEGG metabolic pathways
4. Get genome annotation

Usage Example

Note: Replace <YOUR_SCP_HUB_API_KEY> with your own SCP Hub API Key. You can obtain one from the SCP Platform.

import asyncio
import json
from mcp import ClientSession
from mcp.client.streamable_http import streamablehttp_client
from mcp.client.sse import sse_client

SERVERS = {
    "ncbi-server": "https://scp.intern-ai.org.cn/api/v1/mcp/9/Origene-NCBI",
    "kegg-server": "https://scp.intern-ai.org.cn/api/v1/mcp/5/Origene-KEGG"
}

async def connect(url, transport_type):
    transport = streamablehttp_client(url=url, headers={"SCP-HUB-API-KEY": "<YOUR_SCP_HUB_API_KEY>"})
    read, write, _ = await transport.__aenter__()
    ctx = ClientSession(read, write)
    session = await ctx.__aenter__()
    await session.initialize()
    return session, ctx, transport

def parse(result):
    try:
        if hasattr(result, 'content') and result.content:
            c = result.content[0]
            if hasattr(c, 'text'):
                try: return json.loads(c.text)
                except: return c.text
        return str(result)
    except: return str(result)

async def main():
    
    sessions = {}
    sessions[], _, _ =  connect(, )
    sessions[], _, _ =  connect(, )

    
    
    result_1 =  sessions[].call_tool(, arguments={})
    data_1 = parse(result_1)
    ()

    
    result_2 =  sessions[].call_tool(, arguments={})
    data_2 = parse(result_2)
    ()

    
    result_3 =  sessions[].call_tool(, arguments={})
    data_3 = parse(result_3)
    ()

    
    result_4 =  sessions[].call_tool(, arguments={})
    data_4 = parse(result_4)
    ()

    
    ()

 __name__ == :
    asyncio.run(main())