McpClientManager.py`¶

文件路径¶

verl/tools/utils/mcp_clients/McpClientManager.py

文件概述¶

MCPClientManager 是 MCP（Model Context Protocol）客户端的管理器，负责与 MCP 服务器建立连接、获取可用工具列表、调用远程工具。它是一个全局单例对象（ClientManager），被所有 MCP 工具共享。

通俗理解：MCP 协议类似于一个"工具市场"。MCPClientManager 就是"市场客户端"——它连接到多个工具服务器，查询每个服务器提供了哪些工具，然后在需要时调用对应的工具。

关键代码讲解¶

1. 类定义和全局单例¶

class MCPClientManager:
    rootServerName = "mcpServers"
    initialized = False
    clients = []
    tool_client_mapping = {}
    rate_limiter = None

# 全局单例
ClientManager = MCPClientManager()

使用类属性实现单例模式。所有 MCP 工具共享同一个 ClientManager 实例。

2. 初始化¶

async def initialize(self, config_path, rate_limit: float = 10.0):
    if self.initialized:
        return
    result = self._load_config(config_path)
    servers = result[self.rootServerName]
    exclude_sse_servers = {self.rootServerName: {}}

    for server_name in servers.keys():
        server = servers[server_name]
        if "auth_token" in server:
            transport = SSETransport(
                url=server["url"],
                headers={"Authorization": f"Bearer {server['auth_token']}"}
            )
            client = Client(transport)
            self.clients.append(client)
        else:
            exclude_sse_servers[self.rootServerName][server_name] = server

    if exclude_sse_servers[self.rootServerName]:
        self.clients.append(Client(exclude_sse_servers))

    self.rate_limiter = TokenBucket(rate_limit)
    self.initialized = True

初始化过程： 1. 从 JSON 配置文件加载 MCP 服务器信息。 2. 分两类处理服务器： - 带认证的 SSE 服务器：使用 SSETransport 建立连接（Server-Sent Events 协议）。 - 无认证的服务器：使用 fastmcp.Client 默认方式连接。 3. 初始化令牌桶速率限制器。

3. 获取工具列表¶

async def fetch_tool_schemas(self, tool_selected_list: list[str]) -> list[dict]:
    tool_schemas = []
    for client in self.clients:
        async with client:
            tools = await client.list_tools_mcp()
            for tool in tools.tools:
                if not tool_selected_list:
                    self.tool_client_mapping[tool.name] = client
                    tool_schemas.append(mcp2openai(tool))
                elif tool.name in tool_selected_list:
                    self.tool_client_mapping[tool.name] = client
                    tool_schemas.append(mcp2openai(tool))
    return tool_schemas

遍历所有 MCP 客户端，获取各服务器提供的工具列表： - 如果 tool_selected_list 为空，获取所有工具。 - 如果指定了列表，只获取指定的工具。 - 使用 mcp2openai 将 MCP 格式转换为 OpenAI 格式。 - 建立 tool_client_mapping：工具名 -> 对应客户端的映射。

4. 调用工具¶

async def call_tool(self, tool_name, parameters, timeout):
    # 速率限制
    while not self.rate_limiter.acquire():
        await asyncio.sleep(0.1)

    client = self.get_client_with_tool_name(tool_name)
    async with client:
        return await client.call_tool_mcp(tool_name, parameters)

调用流程： 1. 等待获取速率限制令牌（每 0.1 秒检查一次）。 2. 通过 tool_client_mapping 找到对应的客户端。 3. 使用客户端调用远程工具。

5. 加载配置¶

def _load_config(self, file: str) -> dict[str, Any]:
    try:
        with open(file) as f:
            return json.load(f)
    except FileNotFoundError:
        logger.warning(f'the "{file}" file was not found')
    except Exception:
        logger.error(f'there was an error reading the "{file}" file')
    return {}

配置文件格式示例：

{
    "mcpServers": {
        "search_server": {
            "url": "http://localhost:8080/sse",
            "auth_token": "your_token_here"
        },
        "local_tools": {
            "command": "python",
            "args": ["tool_server.py"]
        }
    }
}

核心类/函数列表¶

类/方法	作用
`MCPClientManager`	MCP 客户端管理器
`initialize`	初始化连接（读取配置、建立客户端）
`fetch_tool_schemas`	获取所有可用工具的 schema
`call_tool`	调用远程工具（带速率限制）
`get_client_with_tool_name`	根据工具名获取对应客户端
`_load_config`	加载 JSON 配置文件
`ClientManager`	全局单例实例

与其他模块的关系¶

被 MCPBaseTool（mcp_base_tool.py）用于执行远程工具调用。
被 tool_registry.py 的 initialize_mcp_tool 函数用于初始化 MCP 工具。
依赖 utils.py 中的 TokenBucket 和 mcp2openai 工具函数。
依赖 fastmcp 库（MCP 协议的 Python 客户端）。

小结¶

MCPClientManager 是 MCP 工具体系的中枢，管理着与所有 MCP 服务器的连接。它实现了服务发现（获取可用工具）、请求路由（将工具调用发送到正确的服务器）和流量控制（速率限制）三大功能。

McpClientManager.py — verl/tools/utils/mcp_clients/McpClientManager.py¶