disease-knowledge-graph

name	disease_knowledge_graph
description	Disease Knowledge Graph - Build disease knowledge graph: OpenTargets targets, drugs, publications, and phenotypes. Use this skill for disease informatics tasks involving get associated targets by disease efoId get associated drugs by target name get publications by drug name get associated phenotypes by disease efoId. Combines 4 tools from 1 SCP server(s).

Disease Knowledge Graph

Discipline: Disease Informatics | Tools Used: 4 | Servers: 1

Description

Build disease knowledge graph: OpenTargets targets, drugs, publications, and phenotypes.

Tools Used

• get_associated_targets_by_disease_efoId from opentargets-server (streamable-http) - https://scp.intern-ai.org.cn/api/v1/mcp/15/Origene-OpenTargets
• get_associated_drugs_by_target_name from opentargets-server (streamable-http) - https://scp.intern-ai.org.cn/api/v1/mcp/15/Origene-OpenTargets
• get_publications_by_drug_name from opentargets-server (streamable-http) - https://scp.intern-ai.org.cn/api/v1/mcp/15/Origene-OpenTargets
• get_associated_phenotypes_by_disease_efoId from opentargets-server (streamable-http) - https://scp.intern-ai.org.cn/api/v1/mcp/15/Origene-OpenTargets

Workflow

1. Get associated targets
2. Get drugs for top target
3. Get publications for top drug
4. Get associated phenotypes

Test Case

Input

{
    "disease_efo": "EFO_0000311"
}

Expected Steps

1. Get associated targets
2. Get drugs for top target
3. Get publications for top drug
4. Get associated phenotypes

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 = {
    "opentargets-server": "https://scp.intern-ai.org.cn/api/v1/mcp/15/Origene-OpenTargets"
}

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():
    # Connect to required servers
    sessions = {}
    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())