客户端接入 MCP Server 实录：基于 Python SDK 的远程工具加载
在 Anthropic 发布的 MCP（Model Context Protocol，模型上下文协议） 标准中，整个生态被清晰地划分为两大核心阵营：
- MCP Server：能力提供方，负责将底层的企业数据库、本地文件系统、Git 代码仓库或外部 SaaS 接口封装为标准化的 Tools 和 Resources；
- MCP Client（客户端 / Agent Host）：智能体编排宿主，负责主动连接一个或多个远程 MCP Server，动态拉取所有可用的工具元数据列表，并在大模型做出工具调用决策时，通过标准协议发起远程调用。
在过去的实践中，官方文档大多以 Claude Desktop 客户端为例进行可视化配置演示。
然而，在企业级后端微服务与自主智能体开发中，我们需要在纯 Python 后端代码中扮演标准的 MCP Client：动态连接分布在跨集群、跨机房的多个 SSE / HTTP 远程 MCP Server，将拉取到的工具无缝注入到 LangGraph、AutoGen 或自研 Agent 编排流中。
本文将手把手实录如何基于官方 Python MCP SDK (mcp) 构建一个生产级、高可用的远程工具加载与执行客户端。
一、MCP Client 核心连接生命周期架构
[ 智能体应用后端 (Python MCP Client Host) ]
│
▼ (1. 建立 SSE 异步会话通道)
┌────────────────────────────────────────────────────────┐
│ 步骤 1: 客户端握手与协议初始化 (Client Session Init) │
│ 发送 initialize 请求，协商协议版本与能力边界 │
└─────────────────────┬──────────────────────────────────┘
│
▼ (2. 动态发现与元数据拉取)
┌────────────────────────────────────────────────────────┐
│ 步骤 2: 工具与资源动态拉取 (tools/list & resources/list)│
│ 拉取所有远端 Tool Name, Description 与 JSON Schema │
└─────────────────────┬──────────────────────────────────┘
│
▼ (3. 转换为大模型标准 Function Calling)
┌────────────────────────────────────────────────────────┐
│ 步骤 3: 注入大模型上下文 (LLM Function Binding) │
│ 大模型输出 ToolCall: {"name": "query_db", "args": {…}}│
└─────────────────────┬──────────────────────────────────┘
│
▼ (4. 远程执行与结果回传)
┌────────────────────────────────────────────────────────┐
│ 步骤 4: 发起 JSON-RPC 远程工具调用 (tools/call) │
│ 跨网络透传至远程 MCP Server，等待并接收标准返回值 │
└────────────────────────────────────────────────────────┘
二、生产级异步 Python MCP 客户端完整实现实操
利用官方 mcp 库中的 sse_client 与 ClientSession，实现支持鉴权与自动工具转换的客户端：
import asyncio
from typing import List, Dict, Any
from mcp import ClientSession
from mcp.client.sse import sse_client
from pydantic import BaseModel
class ProductionMCPClient:
def __init__(self, sse_server_url: str, auth_bearer_token: str = ""):
self.server_url = sse_server_url
self.headers = {"Authorization": f"Bearer {auth_bearer_token}"} if auth_bearer_token else {}
self.session: ClientSession = None
self._exit_stack = None
async def connect_and_discover_tools(self) -> List[Dict[str, Any]]:
"""1. 建立长连接并拉取全量工具定义"""
print(f"【MCP Client】正在连接远程 MCP Server: {self.server_url} …")
# 建立底层 SSE 传输通道
self._sse_context = sse_client(url=self.server_url, headers=self.headers)
streams = await self._sse_context.__aenter__()
# 创建客户端会话
self.session = ClientSession(streams[0], streams[1])
await self.session.__aenter__()
# 握手初始化
await self.session.initialize()
print("【MCP Client】握手成功，开始拉取远程工具元数据…")
# 2. 拉取工具列表
tools_response = await self.session.list_tools()
formatted_tools = []
for tool in tools_response.tools:
# 转换为标准 OpenAI / 大模型兼容的 Function Calling 格式
formatted_tools.append({
"type": "function",
"function": {
"name": tool.name,
"description": tool.description,
"parameters": tool.inputSchema
}
})
print(f" └── 成功加载远程工具: [{tool.name}]")
return formatted_tools
async def invoke_remote_tool(self, tool_name: str, arguments: Dict[str, Any]) -> Any:
"""3. 执行远程工具调用 (tools/call)"""
if not self.session:
raise RuntimeError("MCP 会话未建立，请先调用 connect_and_discover_tools()")
print(f"【MCP Client】发起远程工具执行: {tool_name} | 参数: {arguments}")
# 通过 JSON-RPC 2.0 远程调用
result = await self.session.call_tool(tool_name, arguments=arguments)
# 提取文本或结构化返回值
output_chunks = []
for content in result.content:
if content.type == "text":
output_chunks.append(content.text)
return "\\n".join(output_chunks)
async def close(self):
"""4. 优雅关闭会话与释放连接"""
if self.session:
await self.session.__aexit__(None, None, None)
if hasattr(self, '_sse_context'):
await self._sse_context.__aexit__(None, None, None)
print("【MCP Client】会话已安全关闭。")
三、与大模型编排流端到端集成实战
在真实业务脚本中，MCP Client 能够以极其轻盈的方式嵌入任何编排逻辑：
async def run_agent_workflow():
# 1. 实例化远程 MCP 客户端 (连接我们在上一篇部署的安全 MCP Server)
mcp_client = ProductionMCPClient(
sse_server_url="https://mcp-gateway.company.internal/sse",
auth_bearer_token="prod-secret-mcp-token-2026"
)
try:
# 2. 动态发现可用工具
tools_schema = await mcp_client.connect_and_discover_tools()
# 3. 假设大模型返回了工具调用决策
decision_tool_name = "query_financial_records"
decision_args = {"department": "技术研发部", "year": 2026}
# 4. 执行远程调用
result = await mcp_client.invoke_remote_tool(decision_tool_name, decision_args)
print(f"【最终执行结果】:\\n{result}")
finally:
await mcp_client.close()
# asyncio.run(run_agent_workflow())
四、生产治理避坑指南
在客户端接入 MCP 时，牢记三点：
通过 Python SDK 深度掌控 MCP 客户端生态，让智能体跨越网络物理边界，自由调度全网海量标准化工具能力。
