教你实现一个MCP Client：一个强大的多服务器工具调用客户端（文末附代码链接）

MCP概述

MCP是一个开放协议，它为应用程序向 LLM 提供上下文的方式进行了标准化。你可以将 MCP 想象成 AI 应用程序的 USB-C 接口。就像 USB-C 为设备连接各种外设和配件提供了标准化的方式一样，MCP 为 AI 模型连接各种数据源和工具提供了标准化的接口。

通用架构

MCP 核心采用客户端-服务器架构，主机应用可以连接多个服务器：

• MCP Hosts: 如 Claude Desktop、IDE 或 AI 工具，希望通过 MCP 访问数据的程序
• MCP Clients:维护与服务器一对一连接的协议客户端
• MCP Servers: 轻量级程序，通过标准的 Model Context Protocol提供特定能力
• 本地数据源: MCP 服务器可安全访问的计算机文件、数据库和服务 远程服

务: MCP 服务器可连接的互联网上的外部系统（如通过 APIs）

1. 本文实现的MCP Client

MCP Client是一个基于Python实现的多服务器工具调用客户端。它的主要功能是：

• 支持同时连接多个MCP服务器
• 支持多种传输协议（stdio和SSE）
• 集成OpenAI/OpenRouter接口，实现智能工具调用
• 提供交互式命令行界面

2. 核心特性

2.1 多服务器支持

MCP Client可以同时连接多个MCP服务器，每个服务器可以提供不同的工具集。通过配置文件mcp_server_config.json，可以轻松管理多个服务器连接。

2.2 传输协议

支持两种传输协议：

• stdio：基于标准输入输出的本地进程通信
• SSE（Server-Sent Events）：基于HTTP的服务器推送事件

2.3 智能工具调用

集成OpenAI/OpenRouter API，实现：

• 自动工具选择
• 参数解析和验证
• 多轮工具调用
• 结果整合和展示

3. 实现细节

3.1 初始化和配置

class MCPClient:
def __init__(self):
"""初始化MCP客户端"""
self.session: Optional[ClientSession] = None
self.exit_stack = AsyncExitStack()
self._streams_contexts = {}

# 初始化OpenAI客户端
self._init_openai_client()

# 加载配置
self.config = self._load_config()

# 存储工具和会话信息
self.all_server_tools: Dict[str, List] = {}
self.active_sessions: Dict[str, ClientSession] = {}

3.2 服务器连接管理

MCP Client支持两种类型的服务器连接：

async def _connect_stdio_server(self, server_name: str, server_config: Dict) -> Optional[ClientSession]:
try:
command = server_config.get("command", "python")
args = server_config.get("args", [])
env = server_config.get("env")

server_params = StdioServerParameters(
command=command,
args=args,
env=env
)

stdio_transport = await self.exit_stack.enter_async_context(stdio_client(server_params))
stdio, write = stdio_transport
session = await self.exit_stack.enter_async_context(ClientSession(stdio, write))

return session

async def _connect_sse_server(self, server_name: str, server_config: Dict) -> Optional[ClientSession]:
try:
url = server_config.get("url")
if not url:
logger.warning(f"服务器 '{server_name}' 缺少URL配置")
return None

streams = await self.exit_stack.enter_async_context(sse_client(url=url))
session = await self.exit_stack.enter_async_context(ClientSession(*streams))

return session

3.3 工具调用流程

工具调用的核心流程包括：

def _prepare_tools_for_openai(self) -> List[Dict]:
"""为OpenAI准备工具列表"""
available_tools = []
for server_name, tools in self.all_server_tools.items():
for tool in tools:
available_tools.append({
"type": "function",
"function": {
"name": f"{server_name}:{tool.name}",
"description": f"[{server_name}] {tool.description}",
"parameters": tool.inputSchema
}
})
return available_tools

async def _execute_tool_call(self, tool_call, messages: List[Dict]) -> str:
tool_call_name = tool_call.function.name
tool_args = json.loads(tool_call.function.arguments)

server_name, tool_name = self._parse_tool_name(tool_call_name)

session = self.active_sessions[server_name]

4. 使用示例

# 启动程序
python client.py

# 交互示例

# 查看服务器列表
> servers

# 查看可用工具
> tools

# 执行查询
> 查询: 请帮我查询数据库状态

5. 总结

MCP Client是一个功能强大的多服务器工具调用客户端，它通过：

• 灵活的服务器连接管理
• 智能的工具调用机制
• 友好的交互界面

为用户提供了一个统一的工具调用平台。通过集成OpenAI/OpenRouter API，它能够智能地理解用户需求，自动选择合适的工具，并执行相应的操作。这种设计不仅提高了工具使用的效率，也大大改善了用户体验。

6.代码链接

https://github.com/simpleyue/python_mcp_client/tree/master