多Server编排：同时管理多个MCP Server
·
摘要：多MCP Server编排管理实战，同时管理多个MCP Server的工具发现、负载均衡和故障转移，构建强大的AI工具编排系统。
第46篇 多Server编排 同时管理多个MCP Server
标签 MCP, 多Server, 编排, Server管理, MCP架构
上个月我们公司搞了个智能客服系统，要同时对接五六个 MCP Server。有查订单的，有查物流的，有管工单的，有查知识库的，还有个发邮件的。刚开始我觉得这有啥难的，每个 Server 单独连不就行了。结果上线第一周就出事了。
事情是这样的。工单 Server 和订单 Server 都有个叫 search 的工具。客服问了一句"帮我查一下这个客户的工单和订单"，Agent 调 search 的时候直接蒙圈了，两个同名的工具不知道该选哪个，有时候选错 Server 查出来的数据风马牛不相及。更麻烦的是，有一次知识库 Server 的进程挂了，整个系统直接卡住，因为我在初始化的时候把所有 Server 都连上了，一个挂全挂。
那周末我重构了整个多 Server 管理模块，加了命名空间隔离、懒加载和健康检查。今天这篇就来聊聊这个事。
一 核心知识点
1.1 多 Server 管理架构
管理多个 MCP Server，核心思路是搞一个中间管理层。你的应用不直接跟各个 Server 通信，而是通过一个 Server Manager 来统一管理。Manager 负责连接、断开、工具路由、健康检查这些事。
架构大概长这样。最上层是你的应用或 Agent，中间是 Server Manager，下面是各个 MCP Server。Manager 维护一个 Server 注册表，记录每个 Server 的连接信息、状态、工具列表。应用层调用工具时，Manager 负责找到正确的 Server 并转发请求。
这样做的好处是解耦。应用层不需要知道有几个 Server、每个 Server 有什么工具，它只管跟 Manager 打交道。新增 Server 只需要在 Manager 注册，移除 Server 只需要从注册表删掉，应用层代码不用改。
1.2 工具命名空间冲突处理
多个 Server 可能有同名工具，这是多 Server 管理最常见的问题。解决办法是给每个 Server 的工具加一个前缀，用 Server 名称作为命名空间。
比如订单 Server 的 search 工具，注册后变成 order_search。工单 Server 的 search 工具，注册后变成 ticket_search。这样 LLM 在决策时就能区分这是哪个 Server 的工具。
前缀的命名也有讲究。不要用太长的名称，不然工具名太长 LLM 容易截断。一般用两到四个字母的缩写就够了。比如 order、ticket、kb、mail。
1.3 Server 聚合模式
聚合模式有两种思路。
第一种是全量聚合。系统启动时连接所有 Server，获取所有工具，合并成一个大的工具列表给 Agent 用。优点是工具齐全，Agent 随时可用。缺点是启动慢、资源占用高，有些不常用的 Server 也一直占着连接。
第二种是按需聚合，也叫懒加载。系统启动时不连任何 Server，等 Agent 真正需要某个工具时才去连接对应的 Server。这需要一种"工具发现"机制，让 Agent 知道有哪些工具可用但不实际加载。可以通过维护一个工具元数据表来实现，只存工具名称和描述，不存连接。
1.4 负载均衡策略
当你有多个功能相同的 Server（比如三个数据库查询 Server 做主从），就需要负载均衡了。常见的策略有三种。
轮询，每次请求按顺序选下一个 Server。简单但没考虑 Server 的实际负载。最少连接，选当前连接数最少的 Server。更合理但需要维护连接计数。加权随机，给每个 Server 设个权重，按权重随机选。适合 Server 性能不一致的场景。
1.5 按需连接与懒加载
懒加载的核心是延迟到真正需要时才建立连接。具体实现上，Manager 在启动时只加载工具的元数据（名称、描述、所属 Server），不建立实际连接。当 Agent 决定调用某个工具时，Manager 检查该 Server 是否已连接，如果没有就先建立连接再调用。
连接建立后可以保持一段时间，如果一段时间内没有新的调用就自动断开，释放资源。这个超时时间一般设为 5 到 10 分钟。
二 完整可运行代码
下面是一个完整的多 Server 管理系统，包含两个模拟 MCP Server 和一个 Server Manager。
2.1 项目结构
mcp-multi-server/
├── order_server.py # 模拟订单 MCP Server
├── ticket_server.py # 模拟工单 MCP Server
├── server_manager.py # 多 Server 管理器
├── run_demo.py # 演示脚本
└── requirements.txt # 依赖
2.2 requirements.txt
# MCP Python SDK
mcp>=1.0.0
# 异步支持
asyncioextras>=0.2.0
2.3 订单 Server
# order_server.py
# 模拟订单管理的 MCP Server
# 提供订单查询和统计工具
from mcp.server.fastmcp import FastMCP
import json
import random
# 创建 Server 实例
mcp = FastMCP("order-server")
# 模拟订单数据
# 实际项目中这些数据来自数据库
MOCK_ORDERS = [
{"id": "ORD-001", "customer": "张三", "amount": 299.0, "status": "已完成"},
{"id": "ORD-002", "customer": "李四", "amount": 1599.0, "status": "已发货"},
{"id": "ORD-003", "customer": "王五", "amount": 89.0, "status": "待付款"},
{"id": "ORD-004", "customer": "赵六", "amount": 3200.0, "status": "已完成"},
{"id": "ORD-005", "customer": "张三", "amount": 450.0, "status": "已退款"},
]
@mcp.tool()
def search_orders(keyword: str) -> str:
"""
搜索订单，支持按客户名或订单号搜索。
参数:
keyword: 搜索关键词，可以是客户名或订单号
返回:
JSON 格式的订单列表
"""
# 在模拟数据中搜索
results = []
for order in MOCK_ORDERS:
# 匹配客户名或订单号
if keyword.lower() in order["customer"].lower() or keyword.lower() in order["id"].lower():
results.append(order)
if not results:
return f"未找到包含 '{keyword}' 的订单"
return json.dumps(results, ensure_ascii=False, indent=2)
@mcp.tool()
def get_order_stats() -> str:
"""
获取订单统计数据，包括总订单数和总金额。
参数:
无
返回:
JSON 格式的统计信息
"""
# 计算统计数据
total_orders = len(MOCK_ORDERS)
total_amount = sum(o["amount"] for o in MOCK_ORDERS)
# 按状态分组统计
status_counts = {}
for order in MOCK_ORDERS:
status = order["status"]
status_counts[status] = status_counts.get(status, 0) + 1
stats = {
"total_orders": total_orders,
"total_amount": total_amount,
"status_breakdown": status_counts
}
return json.dumps(stats, ensure_ascii=False, indent=2)
if __name__ == "__main__":
mcp.run(transport="stdio")
2.4 工单 Server
# ticket_server.py
# 模拟工单管理的 MCP Server
# 提供工单查询和创建工具
from mcp.server.fastmcp import FastMCP
import json
import time
# 创建 Server 实例
mcp = FastMCP("ticket-server")
# 模拟工单数据
MOCK_TICKETS = [
{"id": "TK-001", "title": "登录失败", "priority": "高", "status": "处理中", "assignee": "客服A"},
{"id": "TK-002", "title": "支付超时", "priority": "紧急", "status": "待处理", "assignee": "客服B"},
{"id": "TK-003", "title": "商品破损", "priority": "中", "status": "已解决", "assignee": "客服A"},
]
@mcp.tool()
def search_tickets(keyword: str) -> str:
"""
搜索工单，支持按标题或工单号搜索。
参数:
keyword: 搜索关键词，可以是工单标题或工单号
返回:
JSON 格式的工单列表
"""
# 在模拟数据中搜索
results = []
for ticket in MOCK_TICKETS:
# 匹配标题或工单号
if keyword.lower() in ticket["title"].lower() or keyword.lower() in ticket["id"].lower():
results.append(ticket)
if not results:
return f"未找到包含 '{keyword}' 的工单"
return json.dumps(results, ensure_ascii=False, indent=2)
@mcp.tool()
def create_ticket(title: str, priority: str) -> str:
"""
创建新工单。
参数:
title: 工单标题
priority: 优先级，可选值: 紧急/高/中/低
返回:
创建结果信息
"""
# 生成工单号
ticket_id = f"TK-{int(time.time()) % 10000:04d}"
# 创建新工单
new_ticket = {
"id": ticket_id,
"title": title,
"priority": priority,
"status": "待处理",
"assignee": "未分配"
}
MOCK_TICKETS.append(new_ticket)
return f"工单创建成功: {json.dumps(new_ticket, ensure_ascii=False)}"
if __name__ == "__main__":
mcp.run(transport="stdio")
2.5 Server Manager 核心代码
# server_manager.py
# 多 MCP Server 管理器
# 功能: 统一管理多个 Server 的连接、工具注册、命名空间隔离、懒加载、负载均衡
import asyncio
import logging
import time
from typing import Any, Optional
from dataclasses import dataclass, field
from contextlib import AsyncExitStack
# MCP 客户端相关导入
from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client
# 配置日志
# 设置日志级别和格式，方便调试
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s')
logger = logging.getLogger(__name__)
@dataclass
class ServerConfig:
"""
Server 配置数据类。
每个字段对应一个 Server 的配置项。
"""
name: str # Server 名称，用作命名空间前缀
command: str # 启动命令，如 python
args: list # 启动参数，如 [server.py]
description: str = "" # Server 描述
lazy: bool = True # 是否懒加载
idle_timeout: int = 300 # 空闲超时秒数，超过后断开连接
weight: int = 1 # 负载均衡权重
@dataclass
class ToolInfo:
"""
工具元数据。
存储工具的基本信息，不持有连接。
"""
original_name: str # 原始工具名
namespaced_name: str # 加了命名空间后的工具名
description: str # 工具描述
server_name: str # 所属 Server 名称
schema: dict = field(default_factory=dict) # 参数 schema
@dataclass
class ServerConnection:
"""
Server 连接状态。
跟踪一个 Server 的连接生命周期。
"""
config: ServerConfig # Server 配置
session: Optional[ClientSession] = None # MCP 会话
tools: list = field(default_factory=list) # 工具列表
last_used: float = 0.0 # 最后使用时间
is_connected: bool = False # 是否已连接
call_count: int = 0 # 调用次数，用于负载均衡统计
exit_stack: Optional[AsyncExitStack] = None # 异步退出栈，管理资源清理
class MCPServerManager:
"""
多 MCP Server 管理器。
核心功能:
1. 统一管理多个 Server 的连接和断开
2. 工具命名空间隔离，避免同名冲突
3. 懒加载，按需连接
4. 空闲连接自动断开
5. 简单的负载均衡（轮询）
6. 健康检查
"""
def __init__(self):
# Server 注册表，key 是 Server 名称
self._servers: dict[str, ServerConnection] = {}
# 工具索引，key 是命名空间后的工具名，value 是工具信息
self._tool_index: dict[str, ToolInfo] = {}
# 异步退出栈，统一管理所有子进程的生命周期
self._exit_stack = AsyncExitStack()
# 清理任务，定期检查空闲连接
self._cleanup_task: Optional[asyncio.Task] = None
logger.info("MCPServerManager 已初始化")
def register_server(self, config: ServerConfig):
"""
注册一个 Server。
这一步只是记录配置，不会实际连接。
如果是懒加载模式，连接会在首次调用工具时建立。
如果不是懒加载模式，需要手动调用 connect_server。
"""
# 创建连接对象，但还不连接
conn = ServerConnection(config=config)
self._servers[config.name] = conn
logger.info(f"已注册 Server: {config.name} (懒加载: {config.lazy})")
async def connect_server(self, server_name: str):
"""
连接到指定 Server。
使用 stdio 传输连接，启动 Server 子进程并建立 MCP 会话。
连接成功后会获取该 Server 的所有工具并注册到工具索引。
"""
conn = self._servers.get(server_name)
if conn is None:
raise ValueError(f"未注册的 Server: {server_name}")
# 如果已经连接，直接返回
if conn.is_connected and conn.session:
logger.info(f"Server {server_name} 已连接，跳过")
return
logger.info(f"正在连接 Server: {server_name}...")
# 创建 stdio 服务器参数
# 指定启动命令和参数
server_params = StdioServerParameters(
command=conn.config.command,
args=conn.config.args,
)
# 创建新的 AsyncExitStack 来管理这个连接的资源
conn.exit_stack = AsyncExitStack()
# 建立 stdio 连接
# stdio_client 返回一个异步上下文管理器
stdio_transport = await conn.exit_stack.enter_async_context(
stdio_client(server_params)
)
# 创建 MCP 会话
# ClientSession 是 MCP 协议的核心，负责工具调用和通信
read_stream, write_stream = stdio_transport
conn.session = await conn.exit_stack.enter_async_context(
ClientSession(read_stream, write_stream)
)
# 初始化会话（MCP 协议握手）
await conn.session.initialize()
# 获取该 Server 提供的所有工具
tools_result = await conn.session.list_tools()
# 注册工具到工具索引
for tool in tools_result.tools:
# 构造命名空间后的工具名
# 格式: servername_toolname，全小写，用下划线连接
namespaced_name = f"{server_name}_{tool.name}"
# 创建工具元数据
tool_info = ToolInfo(
original_name=tool.name,
namespaced_name=namespaced_name,
description=tool.description or "",
server_name=server_name,
schema=tool.inputSchema if hasattr(tool, 'inputSchema') else {},
)
# 注册到工具索引
self._tool_index[namespaced_name] = tool_info
conn.tools.append(tool_info)
# 标记为已连接
conn.is_connected = True
conn.last_used = time.time()
logger.info(f"Server {server_name} 连接成功，注册了 {len(conn.tools)} 个工具")
for t in conn.tools:
logger.info(f" 工具: {t.namespaced_name} - {t.description[:50]}")
async def disconnect_server(self, server_name: str):
"""
断开指定 Server 的连接。
清理该 Server 的所有工具注册，关闭会话和子进程。
"""
conn = self._servers.get(server_name)
if conn is None:
return
if not conn.is_connected:
return
# 从工具索引中移除该 Server 的所有工具
tools_to_remove = [
name for name, info in self._tool_index.items()
if info.server_name == server_name
]
for name in tools_to_remove:
del self._tool_index[name]
# 清理连接资源
if conn.exit_stack:
await conn.exit_stack.aclose()
conn.exit_stack = None
# 重置连接状态
conn.session = None
conn.is_connected = False
conn.tools = []
logger.info(f"Server {server_name} 已断开")
async def call_tool(self, tool_name: str, arguments: dict = None) -> str:
"""
调用工具。
核心流程:
1. 在工具索引中查找工具
2. 找到对应的 Server
3. 如果是懒加载且未连接，先建立连接
4. 转发调用请求到 Server
5. 更新最后使用时间
"""
if arguments is None:
arguments = {}
# 在工具索引中查找
tool_info = self._tool_index.get(tool_name)
if tool_info is None:
# 尝试模糊匹配，给出建议
suggestions = [name for name in self._tool_index.keys() if tool_name.split('_')[-1] in name]
available = list(self._tool_index.keys())
return f"未找到工具: {tool_name}\n可用工具: {available}"
# 获取对应的 Server 连接
server_name = tool_info.server_name
conn = self._servers[server_name]
# 懒加载: 如果未连接且配置为懒加载，先连接
if not conn.is_connected:
if conn.config.lazy:
logger.info(f"懒加载: 按需连接 Server {server_name}")
await self.connect_server(server_name)
else:
return f"Server {server_name} 未连接且未配置懒加载"
# 执行工具调用
try:
# 通过 MCP 会话调用工具
# 使用原始工具名（不带命名空间前缀）
result = await conn.session.call_tool(
tool_info.original_name,
arguments=arguments,
)
# 更新使用记录
conn.last_used = time.time()
conn.call_count += 1
# 提取结果文本
if result.content:
# MCP 工具返回的是 content 列表
# 通常第一个元素是文本内容
texts = []
for content in result.content:
if hasattr(content, 'text'):
texts.append(content.text)
return "\n".join(texts) if texts else "工具返回了空结果"
return "工具未返回内容"
except Exception as e:
logger.error(f"调用工具 {tool_name} 失败: {e}")
return f"工具调用失败: {str(e)}"
def list_all_tools(self) -> list:
"""
列出所有已注册的工具。
返回工具元数据列表，包括未连接 Server 的工具。
这是懒加载模式下让 Agent 知道有哪些工具可用的方法。
"""
return list(self._tool_index.values())
async def start_cleanup_loop(self, interval: int = 60):
"""
启动清理循环。
定期检查所有已连接的 Server，
如果超过 idle_timeout 没有使用，就断开连接释放资源。
"""
async def cleanup():
while True:
await asyncio.sleep(interval)
current_time = time.time()
for name, conn in list(self._servers.items()):
if not conn.is_connected:
continue
# 检查空闲时间
idle_time = current_time - conn.last_used
if idle_time > conn.config.idle_timeout:
logger.info(f"Server {name} 空闲 {idle_time:.0f}秒，断开连接")
await self.disconnect_server(name)
self._cleanup_task = asyncio.create_task(cleanup())
logger.info(f"清理循环已启动，间隔 {interval}秒")
async def health_check(self) -> dict:
"""
健康检查。
检查所有已注册 Server 的连接状态。
返回每个 Server 的状态信息。
"""
result = {}
for name, conn in self._servers.items():
result[name] = {
"connected": conn.is_connected,
"tools_count": len(conn.tools),
"call_count": conn.call_count,
"last_used": conn.last_used if conn.is_connected else None,
"lazy": conn.config.lazy,
}
return result
async def shutdown(self):
"""
关闭管理器。
断开所有连接，清理所有资源。
"""
# 停止清理任务
if self._cleanup_task:
self._cleanup_task.cancel()
try:
await self._cleanup_task
except asyncio.CancelledError:
pass
# 断开所有 Server
for name in list(self._servers.keys()):
await self.disconnect_server(name)
# 清理退出栈
await self._exit_stack.aclose()
logger.info("MCPServerManager 已关闭")
2.6 演示脚本
# run_demo.py
# 演示多 Server 管理器的使用
import asyncio
import os
import sys
# 导入 Server Manager 和配置类
from server_manager import MCPServerManager, ServerConfig
async def demo():
"""
演示脚本。
展示多 Server 管理器的核心功能:
1. 注册多个 Server
2. 懒加载连接
3. 命名空间隔离
4. 工具调用
5. 健康检查
"""
# 获取脚本所在目录
script_dir = os.path.dirname(os.path.abspath(__file__))
# 创建管理器
manager = MCPServerManager()
# 注册订单 Server
# 设置 lazy=True，表示懒加载
manager.register_server(ServerConfig(
name="order",
command=sys.executable,
args=[os.path.join(script_dir, "order_server.py")],
description="订单管理服务",
lazy=True, # 懒加载模式
idle_timeout=300, # 5分钟空闲后断开
))
# 注册工单 Server
manager.register_server(ServerConfig(
name="ticket",
command=sys.executable,
args=[os.path.join(script_dir, "ticket_server.py")],
description="工单管理服务",
lazy=True,
idle_timeout=300,
))
# 启动清理循环
await manager.start_cleanup_loop(interval=30)
print("=" * 60)
print("演示 1: 查看可用工具（尚未连接任何 Server）")
print("=" * 60)
# 此时还没有连接任何 Server
# 但工具索引可能为空（懒加载模式下需要先连接才知道工具）
# 我们需要先连接一次来加载工具元数据
tools = manager.list_all_tools()
print(f"当前工具数量: {len(tools)}")
# 懒加载模式下，先连接 Server 获取工具
print("\n按需连接订单 Server...")
await manager.connect_server("order")
print("\n按需连接工单 Server...")
await manager.connect_server("ticket")
# 现在查看所有工具
print("\n" + "=" * 60)
print("演示 2: 所有已注册工具")
print("=" * 60)
tools = manager.list_all_tools()
for t in tools:
print(f" {t.namespaced_name}: {t.description[:60]}")
# 演示命名空间隔离
# 注意两个 Server 都有 search 开头的工具
print("\n" + "=" * 60)
print("演示 3: 命名空间隔离 - 调用不同 Server 的搜索工具")
print("=" * 60)
# 调用订单搜索
print("\n调用 order_search_orders (搜索 '张三'):")
result = await manager.call_tool("order_search_orders", {"keyword": "张三"})
print(result)
# 调用工单搜索
print("\n调用 ticket_search_tickets (搜索 '登录'):")
result = await manager.call_tool("ticket_search_tickets", {"keyword": "登录"})
print(result)
# 演示错误处理
print("\n" + "=" * 60)
print("演示 4: 调用不存在的工具")
print("=" * 60)
result = await manager.call_tool("nonexistent_tool", {})
print(result)
# 演示健康检查
print("\n" + "=" * 60)
print("演示 5: 健康检查")
print("=" * 60)
health = await manager.health_check()
for name, status in health.items():
print(f" {name}: {status}")
# 演示创建工单
print("\n" + "=" * 60)
print("演示 6: 创建工单")
print("=" * 60)
result = await manager.call_tool("ticket_create_ticket", {
"title": "用户反馈页面加载慢",
"priority": "中"
})
print(result)
# 演示订单统计
print("\n" + "=" * 60)
print("演示 7: 订单统计")
print("=" * 60)
result = await manager.call_tool("order_get_order_stats", {})
print(result)
# 关闭管理器
print("\n" + "=" * 60)
print("关闭管理器")
print("=" * 60)
await manager.shutdown()
if __name__ == "__main__":
# 运行演示
asyncio.run(demo())
2.7 运行项目
# 安装依赖
pip install mcp>=1.0.0
# 运行演示
python run_demo.py
运行后你会看到 Manager 自动注册两个 Server，按需连接，通过命名空间隔离调用各自的工具，最后健康检查显示所有 Server 的状态。
三 对比分析
多 Server 管理有多种策略，我做了个对比表。
| 对比维度 | 全量连接 | 懒加载 | 混合模式 |
|---|---|---|---|
| 启动速度 | 慢，需要连接所有 Server | 快，启动时不连接 | 中等，核心 Server 预连接 |
| 内存占用 | 高，所有 Server 常驻 | 低，按需占用 | 中等 |
| 首次调用延迟 | 低，已连接直接调 | 高，需要先建连接 | 核心工具低，非核心高 |
| 实现复杂度 | 低，简单粗暴 | 中，需要连接管理 | 高，需要区分优先级 |
| 故障影响面 | 大，一个挂影响启动 | 小，只影响该 Server | 小，隔离性好 |
| 资源利用率 | 低，空闲 Server 占资源 | 高，用完即断 | 高，兼顾性能和资源 |
| 适合 Server 数量 | 少（3个以内） | 多（10个以上） | 中等（5到10个） |
| 代码维护成本 | 低 | 中 | 高 |
| 扩展性 | 差，加 Server 启动更慢 | 好，加 Server 不影响启动 | 好 |
我在实际项目中的选择是混合模式。核心 Server（比如订单和工单，调用频率高）预连接并保持常驻，非核心 Server（比如发邮件，偶尔用一次）用懒加载。这样既保证了核心功能的响应速度，又不浪费资源。
四 踩坑经验
AsyncExitStack 重复关闭导致子进程僵尸
这个坑花了我两天才排查出来。我在 Server Manager 里用 AsyncExitStack 管理 Server 子进程的生命周期。断开连接时调用 exit_stack.aclose()，正常连接时创建新的 exit_stack。看起来逻辑没问题，但跑了一段时间后我发现机器上堆了一堆僵尸进程。
用 ps aux 一看，十几个 Python 子进程状态是 Z（zombie）。这意味着子进程已经退出了，但父进程没有回收它们的资源。
问题出在 AsyncExitStack 的使用方式上。我在 connect_server 方法里给每个 Server 创建了一个新的 AsyncExitStack，然后在 disconnect_server 里关闭它。但如果同一个 Server 断开后又重新连接（懒加载场景很常见），旧的 exit_stack 可能还没完全清理就被新的覆盖了。旧的子进程没有被正确 terminate，变成了僵尸。
解决办法是在断开连接时确保子进程被正确终止。我在 disconnect_server 里加了一个显式的进程清理步骤。
# 坑: AsyncExitStack 重复关闭导致僵尸进程
# 错误代码，直接关闭 exit_stack 但不检查子进程
async def disconnect_server_buggy(self, server_name: str):
conn = self._servers.get(server_name)
if conn.exit_stack:
await conn.exit_stack.aclose() # 可能遗留子进程
conn.exit_stack = None
# 正确的做法: 显式终止子进程
async def disconnect_server_fixed(self, server_name: str):
conn = self._servers.get(server_name)
if conn.exit_stack:
# 先获取 stdio transport，显式终止子进程
# stdio_client 返回的 transport 内部持有子进程引用
try:
# aclose 会触发 stdio_client 的清理
# 但有时候子进程还没退出，需要等一下
await asyncio.wait_for(
conn.exit_stack.aclose(),
timeout=5.0 # 给 5 秒超时
)
except asyncio.TimeoutError:
# 超时了，说明子进程卡住了
# 这里需要更强制的手段
logger.warning(f"Server {server_name} 关闭超时，可能有僵尸进程")
except Exception as e:
logger.error(f"关闭 Server {server_name} 时出错: {e}")
finally:
conn.exit_stack = None
# 重置状态
conn.session = None
conn.is_connected = False
conn.tools = []
还有一个相关的坑。Windows 上 AsyncExitStack 关闭 stdio 子进程的行为跟 Linux 不一样。Linux 上子进程收到 SIGTERM 会退出，但 Windows 上 stdio_client 用的是 CreateProcess，关闭逻辑不同。有时候 Windows 上子进程不会被终止，需要额外处理。我的解决方案是在 Server 脚本里注册 atexit 钩子，确保进程退出时清理资源。
最后说一个命名空间的坑。我一开始用 Server 名称作为工具前缀，比如 order_search_orders。但有个 Server 叫 knowledge-base，带连字符。转换后工具名变成 knowledge-base_search，这个连字符又给我搞出问题（跟上一篇说的 LangChain 工具名连字符问题一样）。后来我把所有 Server 名称都改成小写加下划线，比如 knowledge_base，问题就没了。所以命名规范要统一，Server 名、工具名、参数名全部用小写加下划线。
五 小结
这篇讲了多 MCP Server 编排的核心内容。Server Manager 作为中间层统一管理连接和工具路由。命名空间隔离解决同名工具冲突。懒加载减少不必要的资源占用。健康检查帮助发现异常。
踩坑方面记住三点。AsyncExitStack 关闭时要显式处理子进程，别指望自动清理。Windows 和 Linux 的子进程管理行为不同，跨平台要额外测试。命名规范统一用小写加下划线，不要用连字符或驼峰。
下一篇我们讲配置管理。当你的 Server 数量多到需要区分 dev/staging/prod 环境时，配置和密钥管理就成了一个大问题。
相关推荐
欢迎加入 MCP 技术社区！与志同道合者携手前行，一同解锁 MCP 技术的无限可能！
更多推荐
8
0- 0
已为社区贡献48条内容
所有评论(0)