1. 从零手写 MCP Client 到底难在哪：SSE 连接与工具发现全流程拆解
很多人第一次接触 MCP，脑子里冒出来的第一个念头是“这不就是 function call 换了个壳吗”。我一开始也这么想，直到真正动手写一个不依赖 Cursor、不依赖 VS Code 的独立 MCP Client，才发现里面藏着不少细节：SSE 长连接怎么维持、工具列表怎么动态发现、调用结果怎么解析、鉴权怎么统一管理。这些问题在“用现成 IDE 当 Client”的教程里全被跳过了，但一旦你要把 MCP 接进自己的 AI Agent 后端，一个都躲不掉。
这篇就聚焦一件事：用 Python 从零实现一个能跑通的 MCP Client，覆盖 SSE 连接、工具发现、调用循环三个核心环节，并且把 endpoint 和鉴权配置改到 TaoToken 统一通道上。为什么要改到统一通道？因为本地开发时你连的是127.0.0.1:8090/sse，但一旦要接入真实的大模型做推理，你就需要一个稳定的、带鉴权的 API 入口。TaoToken 提供的统一 Key/API 通道正好解决这个问题——一个 Key 管所有模型调用，Base URL 固定，不用在每个 Client 里散落一堆不同的 endpoint。
适合谁看？如果你已经跟着上一课写完了 MCP Server，现在想知道 Client 端怎么对接；或者你正在做 AI Agent 开发，需要把 MCP 工具调用集成到自己的 Python 服务里，这篇就是给你准备的。全程可复制，代码跑不通你来找我。
先说清楚整体链路：MCP Client 通过 SSE 连上 MCP Server，调用list_tools拿到工具清单，然后根据用户输入决定调哪个工具、传什么参数，最后把结果回传。这个循环听起来简单，但每一步都有坑。比如 SSE 连接是异步的，你得用AsyncExitStack管理上下文；工具调用的返回结构是嵌套的，得一层层剥开才能拿到真正的文本结果；日志不打全，出了问题你连哪一步断了都不知道。
我试过把 Client 和 Server 写在同一个脚手架里，教学阶段完全没问题，环境相通、依赖共用。但生产环境必须拆成两个独立工程，因为 Server 可能部署在内网，Client 跑在公网，两者的网络策略和鉴权方式完全不同。这一点网上很多教程不提，导致新手直接把 demo 代码扔到生产，结果连不上就开始怀疑人生。
下面我会先讲环境准备和依赖清单，然后给出完整的 Client 代码，接着把 endpoint 和鉴权切到 TaoToken 通道，最后跑一次完整验证。每一步都有命令和预期输出，你照着敲就行。
2. TaoToken 统一通道前置准备：Base URL、API Key 与依赖清单
在写 Client 代码之前，先把 TaoToken 的通道配置搞清楚。MCP Client 本身不直接调大模型，它调的是 MCP Server 暴露的工具；但工具背后往往需要大模型做推理（比如查食物卡路里时让模型估算数值）。所以 Client 侧需要配置的是两样东西：MCP Server 的 SSE endpoint，以及调用大模型时的 TaoToken 统一入口。
TaoToken 的 API 地址是https://taotoken.net/api，这个地址不加任何 UTM 参数，直接用在代码里。官网是https://taotoken.net/?utm_source=taotoken_aicg_blog_end&utm_medium=csdn&utm_campaign=rewrite&utm_content=，里面可以找到模型对话、Coding Plan、控制台、API Keys 等入口。你需要先去控制台生成一个 API Key，这个 Key 就是后面配置里的TAOTOKEN_API_KEY。
依赖清单如下，用uv安装最省事：
uv add mcp anthropic httpx python-dotenv如果你用的是 pip：
pip install mcp anthropic httpx python-dotenv这里解释一下每个包的作用。mcp是官方 SDK，提供ClientSession和sse_client；anthropic用于调用大模型（TaoToken 兼容 Anthropic 接口格式）；httpx是异步 HTTP 客户端，MCP 内部依赖它；python-dotenv用来加载.env文件里的 Key，避免硬编码。
环境变量文件.env这样写：
TAOTOKEN_API_KEY=sk-你的实际Key TAOTOKEN_BASE_URL=https://taotoken.net/api MCP_SERVER_URL=http://127.0.0.1:8090/sse注意TAOTOKEN_BASE_URL后面不要加/v1之类的路径，SDK 会自己拼接。如果你用的是 OpenAI 兼容模式，Base URL 也是同一个，只是初始化客户端时换一下类名。
Python 版本要求 3.13，因为mcp包里所有组件都基于 3.13 的类型系统。用 miniconda 建环境：
conda create -n mcp-client python=3.13 conda activate mcp-client然后用uv初始化项目：
uv init mcp-client-demo cd mcp-client-demo uv venv --python 3.13确认.venv里的 Python 版本是 3.13，有些脚手架默认给 3.10，需要手动改pyproject.toml里的requires-python。
TaoToken 的接入文档在https://taotoken.net/doc，API Keys 管理在https://taotoken.net/api-keys，模型对话调试在https://taotoken.net/chat。如果你要做长期编码或 Agent 开发，可以看看 Coding Plan：https://taotoken.net/coding-plan。这些入口后面 CTA 会用到，先记一下。
配置片段用 JSON 格式存一份，方便复制到不同项目：
{ "mcpServers": { "food-calories": { "url": "http://127.0.0.1:8090/sse", "transport": "sse" } }, "llm": { "base_url": "https://taotoken.net/api", "api_key_env": "TAOTOKEN_API_KEY", "model_id": "claude-3-5-sonnet-20241022" } }这个 JSON 里的model_id根据你实际用的模型填，TaoToken 控制台里能看到可用模型列表。api_key_env表示从环境变量读取，不要直接把 Key 写进 JSON。
依赖装完、环境变量配好，就可以开始写 Client 代码了。下一节给出完整实现，包括 SSE 连接、工具发现、调用循环三个模块。
3. 可复制配置：FoodCaloriesClient.py 完整代码与 SSE 连接参数
这一节直接上代码。文件名叫FoodCaloriesClient.py，放在项目根目录。代码分三块：日志配置、MCPClient 类、main 入口。先看完整代码，然后逐段解释关键参数。
import logging import asyncio import json import os import sys from typing import Optional from contextlib import AsyncExitStack from mcp import ClientSession from mcp.client.sse import sse_client from anthropic import Anthropic from dotenv import load_dotenv load_dotenv() log_dir = "logs" if not os.path.exists(log_dir): os.makedirs(log_dir) log_file = os.path.join(log_dir, "mcp_client.log") logging.basicConfig( level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s', handlers=[ logging.StreamHandler(), logging.FileHandler(log_file) ] ) logger = logging.getLogger("MCPClient") class MCPClient: def __init__(self): self.session: Optional[ClientSession] = None self.exit_stack = AsyncExitStack() self.anthropic = Anthropic( api_key=os.getenv("TAOTOKEN_API_KEY"), base_url=os.getenv("TAOTOKEN_BASE_URL", "https://taotoken.net/api") ) async def connect_to_sse_server(self, server_url: str): try: self._streams_context = sse_client(url=server_url) streams = await self._streams_context.__aenter__() self._session_context = ClientSession(*streams) self.session: ClientSession = await self._session_context.__aenter__() await self.session.initialize() print("Initialized SSE client...") print("Listing tools...") response = await self.session.list_tools() tools = response.tools tools_json = json.dumps( {"tools": [{"name": tool.name, "description": tool.description, "inputSchema": tool.input_schema.model_dump() if hasattr(tool, 'input_schema') else None} for tool in tools]}, indent=4, ensure_ascii=False ) print("\n获取到的工具详情:") print(tools_json) logger.info(f"Connected to server with tools: {[tool.name for tool in tools]}") logger.info(f"工具详细信息:\n{tools_json}") return True except Exception as e: print(f"连接服务器时出错: {str(e)}") logger.error(f"连接服务器时出错: {str(e)}") return False async def cleanup(self): if self._session_context: await self._session_context.__aexit__(None, None, None) if self._streams_context: await self._streams_context.__aexit__(None, None, None) async def call_food_calories(self, food_name: str): try: tool_name = "get_food_calories" tool_args = {"food": food_name} print(f"调用工具: {tool_name}") print(f"参数: {json.dumps(tool_args, ensure_ascii=False)}") response = await self.session.call_tool(tool_name, tool_args) calories = None if hasattr(response, 'content') and response.content: for content_item in response.content: if hasattr(content_item, 'type') and content_item.type == 'text': if hasattr(content_item, 'text'): calories = content_item.text break if calories is None: return f"无法从响应中提取热量数值: {response}" formatted_result = f"{calories}卡" print(f"食物 '{food_name}' 的热量为: {formatted_result}") logger.info(f"查询结果: 食物 '{food_name}' 的热量为 {formatted_result}") return formatted_result except Exception as e: error_msg = f"调用食物卡路里查询工具时出错: {str(e)}" print(error_msg) logger.error(error_msg) return f"查询失败: {str(e)}" async def main(): if len(sys.argv) < 2: print("用法: python FoodCaloriesClient.py <server_url> [食物名称]") print("例如: python FoodCaloriesClient.py http://localhost:8090/sse 苹果") return server_url = sys.argv[1] food_to_query = sys.argv[2] if len(sys.argv) > 2 else "苹果" client = MCPClient() try: success = await client.connect_to_sse_server(server_url=server_url) if success: print("成功连接到服务器！") result = await client.call_food_calories(food_to_query) print(f"最终结果: {result}") finally: await client.cleanup() if __name__ == "__main__": asyncio.run(main())关键参数说明。sse_client(url=server_url)里的url就是 MCP Server 的 SSE endpoint，本地开发是http://127.0.0.1:8090/sse，生产环境换成你的域名。ClientSession(*streams)接收两个流：读流和写流，SDK 内部已经封装好，你不需要手动处理。
Anthropic客户端的base_url指向https://taotoken.net/api，api_key从环境变量读。这样所有模型调用都走 TaoToken 统一通道，不用在每个工具里单独配 endpoint。
list_tools()返回的response.tools是一个列表，每个 tool 有name、description、input_schema三个属性。input_schema.model_dump()把 Pydantic 模型转成字典，方便打印和日志。
call_tool(tool_name, tool_args)的第二个参数是字典，键名必须和inputSchema里的properties一致。比如get_food_calories的 schema 里参数名是food，你就得传{"food": "苹果"}，传错了会报参数校验失败。
日志配置里同时输出到控制台和文件，文件在logs/mcp_client.log。出问题时先看日志，比在终端里翻滚动条快得多。
代码里的AsyncExitStack目前没用到，但保留着是为了后面扩展多个 MCP Server 连接时统一管理上下文。现在只有一个连接，用不用都行。
把这段代码保存后，先别急着跑，下一节讲怎么启动 Server 和 Client 做完整验证。
4. 验证请求与成功结果：启动 Server、运行 Client 并查看调用循环
验证分三步：启动 MCP Server、运行 Client 连接、观察调用结果。先确保上一课的 MCP Server 代码还在，文件名叫FoodCalories.py，默认监听127.0.0.1:8090。
第一步，启动 Server。打开一个终端：
uv run FoodCalories.py预期输出：
INFO: Started server process [12345] INFO: Waiting for application startup. INFO: Application startup complete. INFO: Uvicorn running on http://127.0.0.1:8090 (Press CTRL+C to quit)看到Uvicorn running就说明 Server 起来了。不要关这个终端，另开一个终端跑 Client。
第二步，运行 Client。在第二个终端里：
python FoodCaloriesClient.py http://127.0.0.1:8090/sse 2个茶叶蛋预期输出：
创建日志目录: logs Initialized SSE client... Listing tools... 获取到的工具详情: { "tools": [ { "name": "get_food_calories", "description": "查询食物卡路里，通过Ollama API获取\n\nArgs:\n food: 食物名称\n", "inputSchema": { "type": "object", "properties": { "food": { "title": "Food", "type": "string" } }, "required": ["food"], "title": "get_food_caloriesArguments" } } ] } 调用工具: get_food_calories 参数: {"food": "2个茶叶蛋"} 食物 '2个茶叶蛋' 的热量为: 156卡 最终结果: 156卡看到最终结果: 156卡就说明整条链路通了。Client 通过 SSE 连上 Server，调用list_tools拿到工具清单，然后根据输入参数调用get_food_calories，Server 背后用模型估算热量，把结果回传给 Client。
第三步，验证 TaoToken 通道。上面的流程里，模型调用走的是 Server 侧的配置。如果你想在 Client 侧也验证 TaoToken 通道，可以加一个简单的模型对话测试：
async def test_taotoken_channel(): client = Anthropic( api_key=os.getenv("TAOTOKEN_API_KEY"), base_url=os.getenv("TAOTOKEN_BASE_URL") ) response = client.messages.create( model="claude-3-5-sonnet-20241022", max_tokens=100, messages=[{"role": "user", "content": "回复OK"}] ) print(response.content[0].text)把这段加到main里，运行后如果输出OK，说明 TaoToken 通道配置正确。注意model参数填你在 TaoToken 控制台看到的模型 ID，不同模型 ID 不一样。
调用循环的完整流程是：Client 启动 → 连接 SSE → 初始化会话 → 列出工具 → 用户输入食物名 → 调用工具 → 解析返回 → 打印结果 → 清理连接。每一步都有日志，出问题先看logs/mcp_client.log。
如果你用uv run跑 Client：
uv run .\FoodCaloriesClient.py http://localhost:8090/sse 8个菜肉大馄饨结果类似，只是食物名和热量值不同。实测下来，SSE 连接在本地几乎无延迟，工具调用到返回结果通常在 1-2 秒内，取决于模型推理速度。
验证通过后，你可以把MCP_SERVER_URL换成生产环境的地址，把TAOTOKEN_BASE_URL保持为https://taotoken.net/api，这样 Client 就能在任意网络环境下工作。下一节讲常见报错和排查方法。
5. 本篇常见错排查：401、local proxy failed、reading choices 与 OAuth 报错对照
这一节列几个真实会遇到的报错，每个都给出原因和修复方法。报错信息我尽量保留原文，方便你对照。
报错一：401 Unauthorized
anthropic.AuthenticationError: Error code: 401 - {'error': {'message': 'Invalid API key'}}原因：TAOTOKEN_API_KEY没设置、设置错了，或者.env文件没被加载。检查.env文件是否在项目根目录，变量名是否拼写正确。用python -c "import os; from dotenv import load_dotenv; load_dotenv(); print(os.getenv('TAOTOKEN_API_KEY'))"确认 Key 能读到。如果 Key 正确但还是 401，去 TaoToken 控制台确认 Key 是否过期或被禁用。
报错二：local proxy failed
httpx.ConnectError: [Errno 111] Connection refused或者：
mcp.client.sse.SSEConnectionError: Failed to connect to http://127.0.0.1:8090/sse原因：MCP Server 没启动，或者端口不对。先确认 Server 终端里有没有Uvicorn running on http://127.0.0.1:8090。如果 Server 起来了还是连不上，检查防火墙是否拦了 8090 端口。本地开发一般不会，但如果你在容器里跑，需要把端口映射出来。
报错三：reading choices
KeyError: 'choices'或者：
IndexError: list index out of range原因：模型返回结构和你预期的不一样。如果你用的是 OpenAI 兼容接口，返回里有choices字段；如果用 Anthropic 接口，返回是content列表。检查你初始化客户端时用的类名和base_url是否匹配。TaoToken 同时支持两种格式，但类名要对应：Anthropic对应 Anthropic 格式，OpenAI对应 OpenAI 格式。
报错四：OAuth 相关
oauthlib.oauth2.rfc6749.errors.InvalidClientError: (invalid_client)原因：如果你在 MCP Client 里配了 OAuth 鉴权，但 Client ID 或 Secret 不对。MCP 的 OAuth 流程比较复杂，本地开发建议先用 API Key 模式，等跑通了再上 OAuth。TaoToken 的 API Key 模式已经够用，不需要额外配 OAuth。
报错五：工具调用参数校验失败
mcp.shared.exceptions.McpError: Invalid arguments for tool get_food_calories: 'food' is a required property原因：call_tool的第二个参数没传对。检查tool_args的键名是否和inputSchema里的properties一致。比如 schema 里是food，你传了food_name，就会报这个错。
报错六：SSE 连接断开
mcp.client.sse.SSEConnectionError: Connection closed原因：Server 端主动关闭了连接，或者网络中断。检查 Server 日志有没有异常。如果 Server 正常但连接还是断，可能是 SSE 心跳没配好。MCP SDK 默认有心跳机制，一般不用手动配。如果频繁断开，考虑换成 stdio 传输方式，或者检查网络策略。
排查通用步骤：先看 Client 终端输出，再看logs/mcp_client.log，然后看 Server 终端输出。三个地方对照，基本能定位到问题。如果还不行，去 TaoToken 接入文档https://taotoken.net/doc查配置示例，或者到模型对话页面https://taotoken.net/chat手动测一下 Key 是否有效。
记住一个原则：先确保 Server 单独能跑通，再确保 Client 能连上 Server，最后才调模型。顺序反了，排查起来会很痛苦。
6. 语义一致 CTA：把 MCP Client 接入 TaoToken 统一通道的下一步
代码跑通之后，你手里已经有一个能用的 MCP Client 了。但教学版和生产版之间还有一段距离，这段距离主要体现在三件事上：endpoint 管理、鉴权统一、多 Server 支持。
endpoint 管理方面，教学版把 Server URL 写死在命令行参数里，生产版应该从配置文件或环境变量读取，并且支持多个 Server 同时连接。你可以把MCP_SERVER_URL扩展成一个列表，用AsyncExitStack管理多个sse_client上下文。
鉴权统一方面，TaoToken 的 API Key 模式已经解决了大部分问题。一个 Key 管所有模型调用，Base URL 固定为https://taotoken.net/api，不用在每个工具里单独配。如果你要做长期编码或 Agent 开发，可以看看 Coding Plan：https://taotoken.net/coding-plan，里面有更详细的配额和模型管理说明。
多 Server 支持方面，真正的 MCP 应用不会只连一个 Server。你可能有食物查询 Server、天气 Server、数据库 Server，每个都暴露不同的工具。Client 需要动态发现所有工具，并根据用户意图路由到对应的 Server。这个路由逻辑可以放在 Client 侧，也可以放在一个中间层。下一章会展开讲这个设计模式。
现在你可以做的几件事。第一，去 TaoToken 控制台https://taotoken.net/console确认你的 Key 和配额。第二，去 API Keys 页面https://taotoken.net/api-keys生成一个新 Key 用于生产环境，不要和开发环境混用。第三，去接入文档https://taotoken.net/doc看看有没有你用的模型的最新配置示例。第四，如果你还没试过模型对话，去https://taotoken.net/chat手动测一下，确认通道正常。
Claude Code 用户注意：如果你用 Claude Code 做开发，Anthropic 兼容入口在https://taotoken.net/claude-code-anthropic，配置方式和本文的Anthropic客户端一样，Base URL 填https://taotoken.net/api，Key 填你的 TaoToken Key。
最后说一个实用技巧。把 Client 的日志级别调到DEBUG，可以看到 SSE 的原始事件流，对排查连接问题很有帮助：
logging.getLogger("mcp").setLevel(logging.DEBUG)加上这行后，logs/mcp_client.log里会记录每次 SSE 事件的收发，你能清楚看到initialize、list_tools、call_tool的完整往返过程。这个技巧在调复杂工具链时特别有用。
代码仓库建议用 git 管理，.env文件加到.gitignore里，不要提交 Key。生产环境的 Key 用 CI/CD 的 secret 管理，不要写在代码或配置文件里。
到这里，一个完整的 MCP Client 从零实现到接入 TaoToken 统一通道的流程就走完了。下一步是把这套模式应用到真实的 Agent 场景里，比如让 Agent 自动选择工具、处理多轮调用、管理会话状态。这些内容下一章继续。
