拓冰网站定制 
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_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_content里面可以找到模型对话、Coding Plan、控制台、API Keys 等入口。你需要先去控制台生成一个 API Key这个 Key 就是后面配置里的TAOTOKEN_API_KEY。依赖清单如下用uv安装最省事uv add mcp anthropic httpx python-dotenv如果你用的是 pippip install mcp anthropic httpx python-dotenv这里解释一下每个包的作用。mcp是官方 SDK提供ClientSession和sse_clientanthropic用于调用大模型TaoToken 兼容 Anthropic 接口格式httpx是异步 HTTP 客户端MCP 内部依赖它python-dotenv用来加载.env文件里的 Key避免硬编码。环境变量文件.env这样写TAOTOKEN_API_KEYsk-你的实际Key TAOTOKEN_BASE_URLhttps://taotoken.net/api MCP_SERVER_URLhttp://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 python3.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/docAPI Keys 管理在https://taotoken.net/api-keys模型对话调试在https://taotoken.net/chat。如果你要做长期编码或 Agent 开发可以看看 Coding Planhttps://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( levellogging.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_keyos.getenv(TAOTOKEN_API_KEY), base_urlos.getenv(TAOTOKEN_BASE_URL, https://taotoken.net/api) ) async def connect_to_sse_server(self, server_url: str): try: self._streams_context sse_client(urlserver_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]}, indent4, ensure_asciiFalse ) print(\n获取到的工具详情:) print(tools_json) logger.info(fConnected 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_asciiFalse)}) 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_urlserver_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(urlserver_url)里的url就是 MCP Server 的 SSE endpoint本地开发是http://127.0.0.1:8090/sse生产环境换成你的域名。ClientSession(*streams)接收两个流读流和写流SDK 内部已经封装好你不需要手动处理。Anthropic客户端的base_url指向https://taotoken.net/apiapi_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 CTRLC 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_caloriesServer 背后用模型估算热量把结果回传给 Client。第三步验证 TaoToken 通道。上面的流程里模型调用走的是 Server 侧的配置。如果你想在 Client 侧也验证 TaoToken 通道可以加一个简单的模型对话测试async def test_taotoken_channel(): client Anthropic( api_keyos.getenv(TAOTOKEN_API_KEY), base_urlos.getenv(TAOTOKEN_BASE_URL) ) response client.messages.create( modelclaude-3-5-sonnet-20241022, max_tokens100, 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跑 Clientuv 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 Unauthorizedanthropic.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 failedhttpx.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 choicesKeyError: 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 Planhttps://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/apiKey 填你的 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 自动选择工具、处理多轮调用、管理会话状态。这些内容下一章继续。 AI Agent开发大全第二十二课-从0开发一个MCP Client并接入TaoToken统一通道
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_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_content里面可以找到模型对话、Coding Plan、控制台、API Keys 等入口。你需要先去控制台生成一个 API Key这个 Key 就是后面配置里的TAOTOKEN_API_KEY。依赖清单如下用uv安装最省事uv add mcp anthropic httpx python-dotenv如果你用的是 pippip install mcp anthropic httpx python-dotenv这里解释一下每个包的作用。mcp是官方 SDK提供ClientSession和sse_clientanthropic用于调用大模型TaoToken 兼容 Anthropic 接口格式httpx是异步 HTTP 客户端MCP 内部依赖它python-dotenv用来加载.env文件里的 Key避免硬编码。环境变量文件.env这样写TAOTOKEN_API_KEYsk-你的实际Key TAOTOKEN_BASE_URLhttps://taotoken.net/api MCP_SERVER_URLhttp://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 python3.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/docAPI Keys 管理在https://taotoken.net/api-keys模型对话调试在https://taotoken.net/chat。如果你要做长期编码或 Agent 开发可以看看 Coding Planhttps://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( levellogging.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_keyos.getenv(TAOTOKEN_API_KEY), base_urlos.getenv(TAOTOKEN_BASE_URL, https://taotoken.net/api) ) async def connect_to_sse_server(self, server_url: str): try: self._streams_context sse_client(urlserver_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]}, indent4, ensure_asciiFalse ) print(\n获取到的工具详情:) print(tools_json) logger.info(fConnected 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_asciiFalse)}) 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_urlserver_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(urlserver_url)里的url就是 MCP Server 的 SSE endpoint本地开发是http://127.0.0.1:8090/sse生产环境换成你的域名。ClientSession(*streams)接收两个流读流和写流SDK 内部已经封装好你不需要手动处理。Anthropic客户端的base_url指向https://taotoken.net/apiapi_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 CTRLC 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_caloriesServer 背后用模型估算热量把结果回传给 Client。第三步验证 TaoToken 通道。上面的流程里模型调用走的是 Server 侧的配置。如果你想在 Client 侧也验证 TaoToken 通道可以加一个简单的模型对话测试async def test_taotoken_channel(): client Anthropic( api_keyos.getenv(TAOTOKEN_API_KEY), base_urlos.getenv(TAOTOKEN_BASE_URL) ) response client.messages.create( modelclaude-3-5-sonnet-20241022, max_tokens100, 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跑 Clientuv 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 Unauthorizedanthropic.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 failedhttpx.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 choicesKeyError: 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 Planhttps://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/apiKey 填你的 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 自动选择工具、处理多轮调用、管理会话状态。这些内容下一章继续。