MCP客户端使用指南：从单Server到多Server管理，我踩过的坑都在这里了

MCP客户端使用指南：从单Server到多Server管理，我踩过的坑都在这里了

写在前面

上一篇文章把MCP的核心概念讲清楚了，有读者说：“概念我懂了，但具体到代码里，MCP客户端到底怎么用？”

这篇文章就来解决这个问题。我会从最基础的单Server连接开始，一步步讲到多Server管理、错误处理、高级用法，中间穿插我自己在实际使用中踩过的坑。

说实话，MCP客户端的API看起来简单，但真用起来细节不少。我第一次写多Server客户端的时候，光是工具名冲突的问题就调试了一下午。希望这篇文章能帮你少走点弯路。

一、基础用法：四步跑通一个MCP客户端

我们先从最简单的情况开始：连接一个MCP Server，列出它的工具，调用一个工具。

第一步：创建客户端

MCP官方SDK提供了Client类，创建方式很简单：

from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client

# 定义Server参数：用什么命令启动Server
server_params = StdioServerParameters(
command="python",
args=["weather_server.py"],
env=None # 可以传环境变量
)

# 创建stdio传输通道
async with stdio_client(server_params) as (read, write):
# 创建客户端会话
async with ClientSession(read, write) as session:
# 初始化连接
await session.initialize()
print("连接成功！")

这里有几个细节要注意：

• StdioServerParameters里的command是启动Server的命令，args是参数。如果你的Server是一个Python脚本，command就是"python"，args就是脚本路径。
• stdio_client是一个异步上下文管理器，它会帮你启动Server进程并建立通信通道。
• ClientSession也是异步上下文管理器，initialize()方法必须调用，这是MCP协议的握手步骤，不调用的话后续操作都会报错。

我第一次用的时候忘了调用initialize()，结果调list_tools()一直报"未初始化"的错，查了半天才发现是少了这一步。

第二步：列出工具

连接成功后，就可以获取Server提供的工具列表了：

tools_result = await session.list_tools()
for tool in tools_result.tools:
print(f"工具名：{tool.name}")
print(f"描述：{tool.description}")
print(f"参数：{tool.inputSchema}")
print("—")

list_tools()返回的是一个ListToolsResult对象，里面的tools字段是工具列表。每个工具包含name、description、inputSchema三个核心字段。

inputSchema是一个JSON Schema对象，描述了工具需要什么参数。比如天气查询工具的inputSchema可能长这样：

{
"type": "object",
"properties": {
"city": {"type": "string", "description": "城市名称"}
},
"required": ["city"]
}

这个Schema很重要，大模型就是靠它来判断该传什么参数的。你在写MCP Server的时候，一定要把Schema写清楚，否则大模型很容易传错参数。

第三步：调用工具

拿到工具列表之后，就可以调用具体的工具了：

result = await session.call_tool(
name="get_weather",
arguments={"city": "北京"}
)

# 解析返回结果
for content in result.content:
if content.type == "text":
print(f"天气结果：{content.text}")

call_tool()的参数很直观：name是工具名，arguments是参数字典。返回结果是一个CallToolResult对象，里面的content是一个列表，每个元素是一段内容。

为什么content是列表而不是单个字符串？因为MCP支持多模态返回——工具可以同时返回文本、图片、结构化数据等多种内容。目前大部分工具只返回文本，但设计上是支持多模态的。

另外，CallToolResult还有一个isError字段，如果工具执行出错，这个字段会是True，content里会包含错误信息。

第四步：关闭连接

因为用了async with上下文管理器，连接会在退出时自动关闭，不需要手动调用。但如果你不是用上下文管理器的方式，记得调用session.close()来释放资源。

二、多Server管理：真正的挑战在这

单Server的用法很简单，十分钟就能上手。但实际项目中，你的Agent可能需要同时连接好几个MCP Server——天气Server、数据库Server、GitHub Server、文件系统Server……这时候问题就来了。

问题1：怎么同时管理多个连接？

我一开始的做法是写一个列表，把所有session存进去，需要的时候遍历。但这样管理起来很麻烦，尤其是连接断开重连的时候。

后来我封装了一个简单的MCPClientManager类，专门负责多Server的连接管理：

class MCPClientManager:
def __init__(self):
self._sessions = {} # server_name -> ClientSession
self._transports = {} # server_name -> transport

async def connect_server(self, name: str, command: str, args: list):
"""连接一个MCP Server"""
server_params = StdioServerParameters(command=command, args=args)
transport = stdio_client(server_params)
read, write = await transport.__aenter__()
session = ClientSession(read, write)
await session.__aenter__()
await session.initialize()

self._sessions[name] = session
self._transports[name] = transport
print(f"已连接Server：{name}")

async def list_all_tools(self) –> dict:
"""获取所有Server的工具列表"""
all_tools = {}
for server_name, session in self._sessions.items():
result = await session.list_tools()
for tool in result.tools:
# 用 server_name.tool_name 的方式避免重名
full_name = f"{server_name}.{tool.name}"
all_tools[full_name] = {
"server": server_name,
"tool": tool
}
return all_tools

async def call_tool(self, full_name: str, arguments: dict):
"""调用工具，full_name格式为 server_name.tool_name"""
server_name, tool_name = full_name.split(".", 1)
session = self._sessions[server_name]
return await session.call_tool(name=tool_name, arguments=arguments)

这个Manager类做了三件事：

问题2：工具名冲突怎么办？

这是我踩的第一个大坑。我连接了两个Server，一个叫"github"，一个叫"gitlab"，它们都提供了一个叫get_repository的工具。

如果直接用工具名调用，大模型根本不知道该调哪个。我的解决方案是上面代码里的做法：给工具名加上Server前缀，变成github.get_repository和gitlab.get_repository。

但这样又有一个问题：大模型生成工具调用的时候，它不知道要加前缀。你需要在给大模型的工具描述里，把前缀加上去。比如：

# 整理给大模型的工具列表
def format_tools_for_llm(all_tools: dict) –> list:
tools = []
for full_name, info in all_tools.items():
tool = info["tool"]
tools.append({
"type": "function",
"function": {
"name": full_name, # 带前缀的完整名称
"description": tool.description,
"parameters": tool.inputSchema
}
})
return tools

这样大模型看到的工具名就是带前缀的，它生成调用请求的时候也会带上前缀，你再根据前缀路由到对应的Server。

问题3：某个Server挂了怎么办？

多Server环境下，某个Server进程崩溃是常有的事。如果不做处理，整个Agent都会因为一个Server挂掉而瘫痪。

我的做法是在call_tool里加异常捕获和自动重连：

async def call_tool(self, full_name: str, arguments: dict, max_retries: int = 2):
server_name, tool_name = full_name.split(".", 1)

for attempt in range(max_retries + 1):
try:
session = self._sessions.get(server_name)
if session is None:
raise ConnectionError(f"Server {server_name} 未连接")
return await session.call_tool(name=tool_name, arguments=arguments)
except Exception as e:
if attempt < max_retries:
print(f"调用失败（第{attempt+1}次），正在重连：{e}")
await self._reconnect(server_name)
else:
raise RuntimeError(f"工具调用失败，已重试{max_retries}次：{e}")

async def _reconnect(self, server_name: str):
"""重连指定Server"""
# 先关闭旧连接
old_session = self._sessions.get(server_name)
if old_session:
await old_session.close()
# 重新连接（需要保存每个Server的启动参数）
config = self._server_configs[server_name]
await self.connect_server(server_name, config["command"], config["args"])

这样即使某个Server临时挂了，客户端也会自动重连，不会影响其他Server的正常工作。

三、错误处理：这些错误你一定会遇到

用MCP客户端的过程中，我遇到过各种各样的错误。整理一下最常见的几种：

错误1：连接失败

ConnectionRefusedError: [Errno 111] Connection refused

这种错误通常是因为Server没有正常启动。检查一下：

• command和args是否正确
• Server脚本是否有语法错误
• Server依赖是否都安装了

调试技巧：先在终端里手动运行Server命令，看看能不能正常启动。如果手动运行都报错，那MCP客户端肯定连不上。

错误2：工具不存在

Method not found: tools/call

或者调用时返回isError=True，错误信息说工具不存在。

这种情况通常是因为：

• 工具名拼写错误
• 你连的Server不对（多Server时路由错了）
• Server端没有正确注册这个工具

调试方法：先调用list_tools()看看Server到底提供了哪些工具，确认工具名是否正确。

错误3：参数校验失败

Invalid params: 'city' is a required property

这是因为你传的参数不符合工具的JSON Schema。检查一下：

• 必填参数是否都传了
• 参数类型是否正确（比如该传数字的传了字符串）
• 参数名是否拼写正确

我经常犯的错误是把city写成City，大小写不一致导致校验失败。JSON Schema是区分大小写的，一定要注意。

错误4：超时

MCP客户端默认没有超时限制，如果Server端卡住了，客户端会一直等。建议自己加超时：

import asyncio

async def call_with_timeout(session, name, arguments, timeout=30):
try:
result = await asyncio.wait_for(
session.call_tool(name=name, arguments=arguments),
timeout=timeout
)
return result
except asyncio.TimeoutError:
raise TimeoutError(f"工具 {name} 调用超时（{timeout}秒）")

超时时间根据工具的实际情况来定，查天气可能5秒就够了，跑代码分析可能需要60秒。

四、高级用法：不只是调用工具

MCP客户端不只能调用工具，还有一些高级功能值得了解。

资源读取

如果Server提供了Resource，客户端可以读取资源内容：

# 列出所有资源
resources = await session.list_resources()

# 读取指定资源
resource_content = await session.read_resource(
uri="file:///home/user/doc.txt"
)

Resource跟Tool的区别是：Tool是"执行一个操作"，Resource是"读取一份数据"。比如文件系统Server，read_file是Tool，而文件本身是Resource。

进度通知

有些工具执行时间比较长，Server可以在执行过程中发送进度通知。客户端可以通过on_progress回调来接收：

async def on_progress(progress):
print(f"进度：{progress.progress} / {progress.total} – {progress.message}")

result = await session.call_tool(
name="long_running_task",
arguments={},
on_progress=on_progress
)

这个功能在处理长时间任务的时候很有用，比如大文件处理、批量数据分析等。

通知（Notification）

MCP还支持Server主动向客户端发送通知，不需要客户端请求。比如Server可以在有新数据的时候主动通知客户端：

# 客户端注册通知处理器
@session.notification("custom/new_data")
async def handle_new_data(params):
print(f"收到新数据：{params}")

这个功能目前用得还不多，但在事件驱动的场景下很有价值。

五、一个完整的实战例子

最后，我把上面讲的内容整合起来，写一个完整的多Server MCP客户端例子。这个例子同时连接天气Server和文件Server，让大模型可以根据用户问题自动选择工具：

import asyncio
from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client

class MultiServerMCPClient:
def __init__(self):
self.sessions = {}
self.server_configs = {}

async def connect(self, name, command, args):
params = StdioServerParameters(command=command, args=args)
transport = stdio_client(params)
read, write = await transport.__aenter__()
session = ClientSession(read, write)
await session.__aenter__()
await session.initialize()
self.sessions[name] = session
self.server_configs[name] = {"command": command, "args": args}

async def get_all_tools(self):
tools = []
for server_name, session in self.sessions.items():
result = await session.list_tools()
for tool in result.tools:
tools.append({
"name": f"{server_name}.{tool.name}",
"description": tool.description,
"parameters": tool.inputSchema
})
return tools

async def call(self, full_name, arguments, timeout=30):
server_name, tool_name = full_name.split(".", 1)
session = self.sessions[server_name]
return await asyncio.wait_for(
session.call_tool(name=tool_name, arguments=arguments),
timeout=timeout
)

async def main():
client = MultiServerMCPClient()

# 连接两个Server
await client.connect("weather", "python", ["weather_server.py"])
await client.connect("filesystem", "python", ["fs_server.py"])

# 获取所有工具
tools = await client.get_all_tools()
print(f"已加载 {len(tools)} 个工具：")
for t in tools:
print(f" – {t['name']}: {t['description']}")

# 调用天气工具
result = await client.call("weather.get_weather", {"city": "上海"})
print(f"\\n上海天气：{result.content[0].text}")

if __name__ == "__main__":
asyncio.run(main())

这个例子虽然简单，但已经包含了多Server管理、工具收集、统一调用的核心逻辑。你可以在这个基础上加上错误处理、自动重连、大模型集成等功能，变成一个生产级的MCP客户端。

写在最后

MCP客户端的使用，入门很简单，但要做好并不容易。单Server连接四步搞定，多Server管理就需要考虑工具冲突、连接管理、错误处理、超时控制等一堆问题。

我的建议是：不要一上来就自己封装复杂的客户端，先用官方SDK的基础API跑通流程，遇到问题再逐步封装。我自己就是从最简单的单Server开始，慢慢加功能，最后才形成了上面那个Manager类。

另外，MCP协议还在快速演进中，客户端API可能会有变化。建议定期关注官方仓库的更新，及时适配新版本。

你们在使用MCP客户端的时候遇到过什么问题？或者有什么好用的封装技巧？欢迎在评论区分享。

下一篇预告：10.2.3 MCP传输方式详解。MCP不只有stdio一种传输方式，还有HTTP、WebSocket等。我们会深入对比各种传输方式的优缺点，以及在什么场景下该用哪种。