Python之MCPClient实现

import asyncio

from fastmcp import Client

from utils.ConfigManager import config
from utils.LoggerManager import logger
from utils.RSAHelper import pack_rsa_decrypted

# 用户认证信息
user_id = config.get("MCP").get("UID")
user_secret_key = config.get("MCP").get("SECRET_KEY")
# MCP server 信息
mcp_server_url = config.get("MCP").get("SERVER_URL")

# 创建MCP客户端
def get_client():
    # 公钥加密密钥
    decrypted = pack_rsa_decrypted(user_id, user_secret_key)
    mcpConfig = {
        "mcpServers": {
            "server_name": {
                # Remote HTTP/SSE server
                "transport": "http",  # or "sse"
                "url": mcp_server_url,
                "headers": {"Authorization": f"Bearer {decrypted}"},
            }
        }
    }
    client = Client(mcpConfig)
    return client

# 获取MCP服务工具列表
async def list_tools():
    tools = None
    try:
        client = get_client()
        async with client:
            # Basic server interaction
            await client.ping()
            # List available operations
            tools = await client.list_tools()
    except Exception as e:
        logger.error(f"list_tools error: {e}")
    logger.info(f"list_tools: {tools}")
    return tools


if __name__ == "__main__":
    tools = asyncio.run(list_tools())
    print(tools)

# 获取MCP服务工具列表
mcp_tools = asyncio.run(list_tools())

# 构建工具描述给OpenAI
def build_tool_descriptions(tools):
    """将MCP工具转换为OpenAI工具调用格式"""
    openai_tools = []
    for tool in tools:
        openai_tools.append({
            "type": "function",
            "function": {
                "name": tool.name,
                "description": tool.description,
                "parameters": tool.inputSchema  # MCP工具的参数模式
            }
        })
    return openai_tools
openai_tools = build_tool_descriptions(mcp_tools)

# 大语言模型配置
response = client.chat.completions.create(
    model=model,
    messages=[
        {"role": "system", "content": "你是中国移动智慧助手"},
        {"role": "user", "content": "雷军为啥最近都是负面新闻。"},
    ],
    stream=True,
    tools=openai_tools,
    tool_choice="auto",
)
tool_calls_full = []
for chunk in response:
    choice = chunk.choices[0]
    tool_calls = choice.delta.tool_calls
    if tool_calls:
        tool_calls_full.append(tool_calls)
    content1 = choice.delta.content
    print(content1)
print("-----------------")
print(tool_calls_full)