1. 多 MCP server 并发的真实痛点与场景拆解如果你正在用 Python 写 MCP client一开始只连一个 MCP server 时代码通常很清爽一个sse_client、一个ClientSession、一份工具列表模型返回tool_calls就直接调。但只要业务稍微复杂一点比如你同时需要「文件系统工具」「数据库查询工具」「内部知识库检索工具」就会立刻撞上三个问题会话怎么管、工具名冲突怎么办、并发调用怎么不串线。我试过最原始的写法是把每个 server 的连接逻辑复制一遍然后写三个if tool_name xxx分支。结果就是新增一个 server 要改五处代码工具重名直接覆盖AsyncExitStack忘了关导致连接泄漏模型一次返回多个tool_calls时只能串行执行慢得离谱。这篇文章就聚焦这个场景用 Python 开发 MCP client同时连接多个 MCP server做统一会话管理、工具路由和并发调用并且通过 TaoToken 的统一 Key/API 通道接入模型侧让整条链路只需要维护一份凭证。先说清楚这套方案适合谁适合已经跑通过单 server MCP client、想扩展到多 server 的 Python 开发者适合在做 Agent 工具编排、需要把多个 MCP server 的工具聚合成一个工具池的工程同学也适合想用统一 API 通道减少多套 Key 管理成本的团队。核心检索词就是 MCP client、MCP server、Python 多 server 并发下面所有代码都可以直接复制改 URL 就跑。整体思路其实就一句话把所有 server 的 session 收进一个字典key 是工具名value 是 session模型选工具时按名字反查 session 去调用。听起来简单但工程细节全在「连接生命周期」和「并发安全」上。下面从 TaoToken 的前置准备开始一步步把可运行的 client 搭出来。2. TaoToken 统一 Key 接入的前置准备多 server 场景下模型侧的调用会变多每轮对话可能触发多次chat.completions.create还要带tools和parallel_tool_calls。如果每个 server 或每个项目都配一套模型 Key管理成本会迅速上升。TaoToken 在这里的作用是提供统一的 API 通道和 Key让 MCP client 的模型调用部分只认一个 Base URL 和一个 Key工具侧连多少个 server 都不影响模型凭证。你需要先拿到两样东西一个可用的 API Key以及确认 Base URL。访问入口在官网 https://taotoken.net/?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_content 注册后在控制台创建 Key。API 地址是 https://taotoken.net/api 注意这个地址不带任何查询参数直接作为 OpenAI SDK 的base_url使用。具体操作路径进入控制台后找到 API Keys 页面deep linkhttps://taotoken.net/console/api-keys?utm_sourcetaotoken_aicg_blog_endutm_contentapi_keysutm_campaignrewrite 新建一个 Key 并复制保存。然后确认你要用的模型 ID可以在模型对话页面deep linkhttps://taotoken.net/model-chat?utm_sourcetaotoken_aicg_blog_endutm_contentmodel_chatutm_campaignrewrite 先手动发一条消息验证通道是否正常。如果你后续要做长期编码或 Agent 类任务可以了解 Coding Plandeep linkhttps://taotoken.net/coding-plan?utm_sourcetaotoken_aicg_blog_endutm_contentcoding_planutm_campaignrewrite 它更适合高频调用场景。这里要强调一个工程习惯把模型凭证和 MCP server 地址分开配置。模型凭证来自 TaoTokenMCP server 地址是你自己的服务两者解耦后换模型通道不影响工具连接加 server 也不影响模型 Key。建议用环境变量管理避免硬编码export TAOTOKEN_API_KEYsk-你的Key export TAOTOKEN_BASE_URLhttps://taotoken.net/api export MCP_SERVER_URLShttp://127.0.0.1:8001/sse,http://127.0.0.1:8002/sse依赖方面需要安装 OpenAI SDK 和 MCP 官方 Python 包pip install openai mcp确认版本能正常 importimport openai from mcp.client.sse import sse_client from mcp.client.session import ClientSession print(openai.__version__)如果这一步报ModuleNotFoundError说明包没装全先解决依赖再往下走。前置准备做完模型侧就只有一个 Key 和一个 Base URL接下来进入多 server 注册的核心实现。3. 多 server 注册与工具路由的可复制配置这一节是全文的技术核心。目标是把多个 MCP server 的连接、工具收集、工具到 session 的映射全部收敛到一段可复制的代码里。先给一个配置文件片段把 server 列表和模型参数外置方便你按项目改。mcp_config.json{ model: gpt-4o-mini, base_url: https://taotoken.net/api, api_key_env: TAOTOKEN_API_KEY, servers: [ { name: filesystem, url: http://127.0.0.1:8001/sse }, { name: database, url: http://127.0.0.1:8002/sse }, { name: knowledge, url: http://127.0.0.1:8003/sse } ] }对应的 Python 加载与 client 初始化import json import os from openai import OpenAI with open(mcp_config.json, r, encodingutf-8) as f: config json.load(f) client OpenAI( api_keyos.environ[config[api_key_env]], base_urlconfig[base_url], ) MODEL config[model] SERVER_URLS [s[url] for s in config[servers]]接下来是多 server 会话管理的核心。关键点有三个用AsyncExitStack统一管理所有异步资源用tool_session_map记录工具名到 session 的映射工具重名时直接报错而不是静默覆盖。import json import logging from contextlib import AsyncExitStack from typing import Dict from mcp.client.sse import sse_client from mcp.client.session import ClientSession from openai.types.chat import ChatCompletionToolParam logger logging.getLogger(__name__) async def build_tool_pool(server_urls, stack: AsyncExitStack): available_tools [] tool_session_map: Dict[str, ClientSession] {} for server_url in server_urls: read, write await stack.enter_async_context( sse_client(urlserver_url) ) session: ClientSession await stack.enter_async_context( ClientSession(read, write) ) await session.initialize() tools await session.list_tools() logger.info(fMCP Server Tools {server_url} {tools}) for tool in tools.tools: if tool.name in tool_session_map: logger.error(f重复名称的工具 {tool.name}) raise ValueError(f重复名称的工具 {tool.name}) tool_session_map[tool.name] session available_tools.append({ type: function, function: { name: tool.name, description: tool.description, parameters: tool.inputSchema, }, }) return available_tools, tool_session_map这段代码里stack.enter_async_context会把每个sse_client和ClientSession都登记进去退出async with时统一关闭避免连接泄漏。tool_session_map是路由表模型返回工具名后直接查表拿 session。工具重名直接抛错这是刻意的设计——多 server 场景下静默覆盖会导致调用到错误的 server排查成本极高。并发调用部分模型侧开启parallel_tool_callsTrueclient 侧用asyncio.gather并发执行多个tool_callsimport asyncio async def execute_tool_calls(tool_calls, tool_session_map): async def run_one(tool_call): function tool_call.function tool_name function.name session tool_session_map.get(tool_name) if session is None: raise ValueError(f模型返回的工具不存在 {tool_name}) result await session.call_tool( tool_name, json.loads(function.arguments) ) if result.isError: raise ValueError(f调用 MCP server tool {tool_name} 返回失败) return tool_call.id, result.content return await asyncio.gather(*(run_one(tc) for tc in tool_calls))到这里多 server 注册、工具路由、并发执行三件套就齐了。配置片段和代码都可以直接复制改mcp_config.json里的 URL 即可适配你自己的 server。4. 连接验证与多 server 并行调用成功结果代码写完必须验证否则你不知道是连接没通、工具没注册还是模型没选对工具。验证分三步先验证单 server 连接再验证工具池聚合最后验证并行调用。第一步单独验证每个 server 能连上并列出工具import asyncio from contextlib import AsyncExitStack async def check_single_server(url): async with AsyncExitStack() as stack: read, write await stack.enter_async_context(sse_client(urlurl)) session await stack.enter_async_context(ClientSession(read, write)) await session.initialize() tools await session.list_tools() print(f{url} - {[t.name for t in tools.tools]}) asyncio.run(check_single_server(http://127.0.0.1:8001/sse))如果这一步报连接错误先确认 server 是否启动、端口是否对、SSE 路径是否为/sse。确认每个 server 单独可用后再跑工具池聚合async def check_tool_pool(): async with AsyncExitStack() as stack: tools, mapping await build_tool_pool(SERVER_URLS, stack) print(f聚合工具数{len(tools)}) for name, session in mapping.items(): print(f{name} - {id(session)}) asyncio.run(check_tool_pool())预期输出是聚合工具数等于所有 server 工具之和且不同工具映射到不同 session 对象。如果工具数偏少检查是否有重名被拦截。第三步是完整的并行调用验证。构造一个需要同时用到两个 server 工具的问题开启parallel_tool_callsTrueasync def query_mcp_server(question: str, parallel: bool True): async with AsyncExitStack() as stack: available_tools, tool_session_map await build_tool_pool( SERVER_URLS, stack ) messages [ {role: system, content: 仅使用 roletool 相关数据进行分析}, {role: user, content: question}, ] while True: response client.chat.completions.create( modelMODEL, messagesmessages, toolsavailable_tools, parallel_tool_callsparallel, ) assistant_output response.choices[0].message if assistant_output.content is None: assistant_output.content if ( response.choices[0].finish_reason ! tool_calls or not assistant_output.tool_calls ): print(f最终结果{assistant_output.content}) return assistant_output.content messages.append(assistant_output) results await execute_tool_calls( assistant_output.tool_calls, tool_session_map ) for tool_call_id, content in results: messages.append({ role: tool, tool_call_id: tool_call_id, content: content, }) asyncio.run(query_mcp_server(同时查询文件列表和数据库表结构))成功时你会看到日志里先打印各 server 的工具列表然后模型返回多个tool_callsasyncio.gather并发执行最后模型汇总结果输出。实测下来两个 server 各一个工具并行调用耗时接近单个调用而不是两倍。这就是多 server 并发的价值。5. 本篇常见报错与排查对照多 server 场景的报错比单 server 更隐蔽因为问题可能出在连接、路由、模型返回任意一环。下面按真实报错对照排查。401 Unauthorized模型侧报错通常是 TaoToken Key 没读到或写错。检查os.environ[TAOTOKEN_API_KEY]是否存在Base URL 是否为https://taotoken.net/api。注意 Base URL 不要带多余路径OpenAI SDK 会自己拼/chat/completions。local proxy failed / connection refusedMCP server 侧报错说明sse_client连不上。确认 server 进程在跑、端口对、URL 路径是/sse。多 server 时逐个用第 4 节的check_single_server验证别一次性全连。reading choices 相关报错模型返回结构不符合预期常见于response.choices为空或finish_reason异常。检查请求是否带了tools以及模型是否支持 function calling。如果模型不支持会直接返回文本而不是tool_calls。OAuth / 鉴权类报错如果某个 MCP server 需要额外鉴权sse_client需要传 headers。多 server 时每个 server 的鉴权方式可能不同建议在配置里为每个 server 单独加headers字段而不是全局共用。工具重名报错重复名称的工具这是设计内的保护。解决办法有两个给 server 的工具加前缀在 server 侧改或者在 client 侧注册时重命名。推荐前者因为模型看到的工具名更语义化。模型返回的工具不存在说明模型返回了工具池里没有的名字。常见原因是available_tools和tool_session_map不同步或者多轮对话里 messages 混入了旧工具。检查build_tool_pool是否每次请求都重新构建以及 messages 是否干净。并发调用结果串线如果多个tool_calls的结果对应错了tool_call_id检查execute_tool_calls是否用asyncio.gather并按返回顺序组装。每个结果必须带自己的tool_call_id不能共用。排查顺序建议先确认模型通道401 类再确认 server 连接refused 类最后确认工具路由重名/不存在类。这样能最快定位问题层。6. 长期编码与 Agent 场景的接入建议多 server MCP client 跑通后下一步通常是把它用到长期编码或 Agent 任务里。这时候有几个工程建议值得提前考虑。第一把build_tool_pool的结果缓存起来不要每轮对话都重连所有 server。连接建立和initialize有开销多 server 时更明显。可以在进程启动时建一次连接池请求时复用 session只在 server 变更时重建。第二工具路由表建议加一层命名空间。比如filesystem.read_file、database.query这样即使不同 server 有同名工具也不会冲突模型也更容易理解工具归属。实现上可以在注册时给tool.name加 server 前缀调用时再剥掉。第三模型侧凭证统一走 TaoToken工具侧 server 地址各自配置。这样换模型、加 server 互不影响。如果你要做的是长期编码类 Agent可以了解 Coding Plandeep linkhttps://taotoken.net/coding-plan?utm_sourcetaotoken_aicg_blog_endutm_contentcoding_planutm_campaignrewrite 如果只是验证模型通道用模型对话页面deep linkhttps://taotoken.net/model-chat?utm_sourcetaotoken_aicg_blog_endutm_contentmodel_chatutm_campaignrewrite 先手动测接入细节查文档deep linkhttps://taotoken.net/doc?utm_sourcetaotoken_aicg_blog_endutm_contentdocutm_campaignrewrite Key 管理在 API Keys 页面deep linkhttps://taotoken.net/console/api-keys?utm_sourcetaotoken_aicg_blog_endutm_contentapi_keysutm_campaignrewrite 。第四并发调用要加超时和异常隔离。asyncio.gather默认一个失败全失败可以用return_exceptionsTrue让单个工具失败不影响其他工具最后统一汇总错误给模型。最后提醒一个容易忽略的点AsyncExitStack的生命周期要和请求生命周期对齐。如果你在长驻进程里复用连接就别用async with包整个请求而是手动管理 stack 的aclose。这一点在从单 server 迁移到多 server 时最容易踩坑。
)
