SCP-PorousKG

				
					# Python
----------
import asyncio
import json
from mcp.client.streamable_http import streamablehttp_client
from mcp.client.session import ClientSession

class MultiDomainKGClient:
    
    def __init__(self, server_url: str = "https://scp.intern-ai.org.cn/api/v1/mcp/37/SciGraph"):
        self.server_url = server_url
        self.session = None
        
    async def connect(self):
        """建立连接并初始化会话"""
        print(f"\n{'='*80}")
        print("连接到 SciGraph SCP Server")
        print(f"{'='*80}")
        print(f"服务器地址: {self.server_url}")
        
        try:
            self.transport = streamablehttp_client(
                url=self.server_url,
                headers={"SCP-HUB-API-KEY": "sk-xxx"}
            )
            self.read, self.write, self.get_session_id = await self.transport.__aenter__()
            
            self.session_ctx = ClientSession(self.read, self.write)
            self.session = await self.session_ctx.__aenter__()
            
            await self.session.initialize()
            session_id = self.get_session_id()
            
            print(f"✓ 连接成功")
            print(f"✓ 会话ID: {session_id}")
            print(f"{'='*80}\n")
            return True
            
        except Exception as e:
            print(f"✗ 连接失败: {e}")
            import traceback
            traceback.print_exc()
            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)
            print("\n✓ 已断开连接\n")
        except Exception as e:
            print(f"✗ 断开连接时出错: {e}")
    
    async def list_tools(self):
        """列出所有可用工具"""
        tools_list = await self.session.list_tools()
        print(f"\n可用工具 (共{len(tools_list.tools)}个):\n")
        for i, tool in enumerate(tools_list.tools, 1):
            print(f"{i:2d}. {tool.name}")
            if tool.description:
                desc_line = tool.description.split('\n')[0]
                print(f"    {desc_line}")
        return tools_list.tools
    
    def parse_result(self, result):
        """解析 MCP 工具调用结果"""
        try:
            if hasattr(result, 'content') and result.content:
                content = result.content[0]
                if hasattr(content, 'text'):
                    return json.loads(content.text)
            return str(result)
        except Exception as e:
            return {"error": f"解析结果失败: {e}", "raw": str(result)}