1. 从一次“查天气”说起MCP 生命周期到底在解决什么问题MCPModel Context Protocol模型上下文协议是什么一句话它是一套让大模型用统一方式“伸手”去拿外部数据和调用外部工具的协议。能做什么把“模型不知道的实时信息”和“模型不该自己瞎编的操作”交给外部服务处理。适合谁正在做 AI Agent、智能客服、IDE 插件、企业内部助手的开发者尤其是被 Function Calling 各家格式折磨过的人。我拿一个最小场景切入用户在聊天框里问“上海今天天气怎么样适合出门吗”。模型训练数据里没有今天的天气它必须调用一个get_weather工具拿到“上海多云27°C”再根据结果推荐活动。整个过程里MCP 要经历四个阶段初始化握手、能力协商列出有哪些工具和资源、工具调用、会话关闭。这四个阶段合起来就是 MCP 生命周期。很多人第一次接触 MCP 会把它和 Function Calling 混为一谈。区别在于Function Calling 是“应用层预先决定给模型哪些函数”而 MCP 是“模型基于上下文自主推理该调哪个工具”工具的实现细节被封装在独立的 MCP Server 里对模型透明。这意味着你新增一个工具只要符合 MCP 协议标准模型侧代码一行都不用改。这篇 DEMO 我会用 TaoToken 统一 Key 作为模型通道把 MCP Server、MCP Client、MCP Host 三段代码串起来让你能亲手跑通一次完整的工具调用并在日志里看到生命周期每个阶段的真实输出。TaoToken 在这里的作用是一个 Key 就能切换不同模型省去为每个模型单独配 Key 的麻烦特别适合做多模型接入验证。2. TaoToken 前置准备统一 Key 与 API 通道配置在写 MCP 代码之前先把模型通道打通。TaoToken 的定位是统一 API 通道你注册后拿到一个 Key就能通过兼容 OpenAI 格式的接口调用多种模型。官网入口是 https://taotoken.net/?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_content API 基址是 https://taotoken.net/api 。第一步去控制台创建 API Key。打开 https://taotoken.net/console?utm_sourcetaotoken_aicg_blog_endutm_contentconsoleutm_campaignrewrite 在 API Keys 页面点新建复制生成的 Key。这个 Key 后面会同时用在 MCP Host 的 LLM 调用里。第二步确认你要用的模型 ID。在模型对话页面 https://taotoken.net/models?utm_sourcetaotoken_aicg_blog_endutm_contentmodelsutm_campaignrewrite 可以看到当前支持的模型列表选一个你熟悉的比如gpt-4o-mini或claude-3-5-sonnet。记下这个 Model ID配置里要用。第三步把 Key 和 Base URL 写进环境变量避免硬编码进代码。在项目根目录建一个.env文件TAOTOKEN_API_KEYsk-你的Key TAOTOKEN_BASE_URLhttps://taotoken.net/api TAOTOKEN_MODELgpt-4o-mini然后在 Python 里用python-dotenv读取。如果你不想装额外依赖也可以直接在 shell 里 exportexport TAOTOKEN_API_KEYsk-你的Key export TAOTOKEN_BASE_URLhttps://taotoken.net/api export TAOTOKEN_MODELgpt-4o-mini这里有个容易踩的坑Base URL 末尾不要带/v1TaoToken 的兼容层会自动补全路径。如果你手动拼成https://taotoken.net/api/v1/chat/completions反而可能 404。正确的请求地址是https://taotoken.net/api/chat/completions。另外MCP Server 本身不需要 TaoToken Key它只负责提供工具。Key 只用在 MCP Host 调用 LLM 的那一步。这个分工要理清楚否则你会以为 Server 也要配 Key。如果你打算长期跑编码类 Agent可以了解下 Coding Planhttps://taotoken.net/coding-plan?utm_sourcetaotoken_aicg_blog_endutm_contentcoding-planutm_campaignrewrite 它针对高频编码场景做了额度优化。接入文档在 https://taotoken.net/doc?utm_sourcetaotoken_aicg_blog_endutm_contentdocutm_campaignrewrite 遇到参数问题可以先查这里。3. 可复制配置MCP Server、Client、Host 三段代码这一节是全文核心我把三段代码完整贴出来你复制就能跑。先装依赖pip install mcp openai python-dotenv注意这里用openai库而不是requests因为 TaoToken 兼容 OpenAI 格式用官方 SDK 更省事也方便你以后换模型。3.1 MCP Server注册两个工具新建mcp_server_demo.pyfrom mcp.server.fastmcp import FastMCP import asyncio mcp FastMCP(nameweather-demo, host0.0.0.0, port1234) mcp.tool(nameget_weather, description获取指定城市的天气信息) async def get_weather(city: str) - str: weather_data { 北京: 北京晴25°C, 上海: 上海多云27°C, 广州: 广州小雨30°C } return weather_data.get(city, f{city}天气信息未知) mcp.tool(namesuggest_activity, description根据天气描述推荐适合的活动) async def suggest_activity(condition: str) - str: if 晴 in condition: return 天气晴朗推荐你去户外散步或运动。 elif 多云 in condition: return 多云天气适合逛公园或咖啡馆。 elif 雨 in condition: return 下雨了建议你在家阅读或看电影。 else: return 建议进行室内活动。 async def main(): print(启动 MCP Server: http://127.0.0.1:1234) await mcp.run_sse_async() if __name__ __main__: asyncio.run(main())这段代码用FastMCP装饰器注册了两个工具。run_sse_async()会启动一个 SSE 服务监听 1234 端口。启动后你会看到启动 MCP Server: http://127.0.0.1:1234。3.2 MCP Client连接 Server 并列出能力新建mcp_client_demo.pyimport asyncio from mcp.client.session import ClientSession from mcp.client.sse import sse_client class WeatherMCPClient: def __init__(self, server_urlhttp://127.0.0.1:1234/sse): self.server_url server_url self._sse_context None self._session None async def __aenter__(self): self._sse_context sse_client(self.server_url) self.read, self.write await self._sse_context.__aenter__() self._session ClientSession(self.read, self.write) await self._session.__aenter__() await self._session.initialize() return self async def __aexit__(self, exc_type, exc_val, exc_tb): if self._session: await self._session.__aexit__(exc_type, exc_val, exc_tb) if self._sse_context: await self._sse_context.__aexit__(exc_type, exc_val, exc_tb) async def list_tools(self): return await self._session.list_tools() async def list_resources(self): return await self._session.list_resources() async def call_tool(self, name, arguments): return await self._session.call_tool(name, arguments) async def main(): async with WeatherMCPClient() as client: print(成功连接 MCP Server) tools await client.list_tools() print(\n可用工具:) print(tools) resources await client.list_resources() print(\n可用资源:) print(resources) print(\n调用 get_weather 工具(city上海)...) result await client.call_tool(get_weather, {city: 上海}) print(\n工具返回:) for item in result.content: print( -, item.text) if __name__ __main__: asyncio.run(main())__aenter__里做了三件事建立 SSE 通道、创建 ClientSession、调用initialize()完成握手。这就是生命周期的初始化阶段。list_tools()是能力协商阶段call_tool()是工具调用阶段__aexit__是会话关闭阶段。3.3 MCP Host串起 LLM 和工具新建mcp_host_demo.pyimport asyncio import json import re import os from openai import OpenAI from dotenv import load_dotenv from mcp_client_demo import WeatherMCPClient load_dotenv() client_llm OpenAI( api_keyos.getenv(TAOTOKEN_API_KEY), base_urlos.getenv(TAOTOKEN_BASE_URL) ) MODEL_ID os.getenv(TAOTOKEN_MODEL, gpt-4o-mini) def extract_json_from_reply(reply: str): if isinstance(reply, dict): return reply if isinstance(reply, str): reply re.sub(r^(?:json)?|$, , reply.strip(), flagsre.IGNORECASE).strip() for _ in range(3): try: parsed json.loads(reply) if isinstance(parsed, dict): return parsed else: reply parsed except Exception: break return reply async def main(): client WeatherMCPClient() await client.__aenter__() tools await client.list_tools() resources await client.list_resources() tool_names [t.name for t in tools.tools] tool_descriptions \n.join(f- {t.name}: {t.description} for t in tools.tools) resource_descriptions \n.join(f- {r.uri} for r in resources.resources) while True: user_input input(\n请输入你的问题输入 exit 退出\n ) if user_input.lower() in (exit, 退出): break system_prompt ( 你是一个智能助手拥有以下工具和资源可以调用\n\n f工具列表\n{tool_descriptions or 无}\n\n f资源列表\n{resource_descriptions or 无}\n\n 请优先调用可用的 Tool 或 Resource而不是 llm 内部生成。 仅根据上下文调用工具不传入不需要的参数进行调用\n 如果需要请以 JSON 返回 tool_calls格式如下\n {tool_calls: [{name: get_weather, arguments: {city: 北京}}]}\n 如无需调用工具返回{\tool_calls\: null} ) messages [ {role: system, content: system_prompt}, {role: user, content: user_input} ] final_reply while True: response client_llm.chat.completions.create( modelMODEL_ID, messagesmessages ) reply response.choices[0].message.content print(f\nLLM 回复\n{reply}) parsed extract_json_from_reply(reply) if isinstance(parsed, str): final_reply parsed break tool_calls parsed.get(tool_calls) if not tool_calls: final_reply parsed.get(content, ) break for tool_call in tool_calls: tool_name tool_call[name] arguments tool_call[arguments] if tool_name not in tool_names: raise ValueError(f工具 {tool_name} 未注册) print(f调用工具 {tool_name} 参数: {arguments}) result await client.call_tool(tool_name, arguments) tool_output result.content[0].text print(f工具 {tool_name} 返回{tool_output}) messages.append({ role: tool, name: tool_name, content: tool_output }) print(f\n最终回复{final_reply}) await client.__aexit__(None, None, None) if __name__ __main__: asyncio.run(main())这里的关键点client_llm用的是 TaoToken 的 Base URL 和 KeyModel ID 从环境变量读。messages里追加role: tool的消息就是把工具结果回传给模型。整个循环直到模型返回纯文本才结束。如果你用 Claude Code 或 Cline 这类工具配置方式类似需要填三件套Base URL 填https://taotoken.net/apiKey 填你的 TaoToken KeyModel ID 填你选的模型。Claude Code 的接入文档在 https://taotoken.net/claude-code?utm_sourcetaotoken_aicg_blog_endutm_contentclaude-codeutm_campaignrewrite 。4. 验证请求生命周期各阶段日志与成功结果先启动 Serverpython mcp_server_demo.py看到启动 MCP Server: http://127.0.0.1:1234就说明初始化成功。这个阶段对应生命周期的“初始化握手”Server 在 1234 端口等待 SSE 连接。另开一个终端先单独测 Clientpython mcp_client_demo.py你应该看到成功连接 MCP Server 可用工具: metaNone nextCursorNone tools[Tool(nameget_weather, description获取指定城市的天气信息, inputSchema{...}), Tool(namesuggest_activity, ...)] 可用资源: metaNone nextCursorNone resources[] 调用 get_weather 工具(city上海)... 工具返回: - 上海多云27°C这段日志覆盖了三个生命周期阶段成功连接是初始化可用工具是能力协商工具返回是工具调用。可用资源为空是因为我们没注册 resource不影响 DEMO。现在跑完整的 Hostpython mcp_host_demo.py输入“上海今天天气怎么样适合出门吗”你会看到类似输出LLM 回复 {tool_calls: [{name: get_weather, arguments: {city: 上海}}]} 调用工具 get_weather 参数: {city: 上海} 工具 get_weather 返回上海多云27°C LLM 回复 {tool_calls: [{name: suggest_activity, arguments: {condition: 多云}}]} 调用工具 suggest_activity 参数: {condition: 多云} 工具 suggest_activity 返回多云天气适合逛公园或咖啡馆。 LLM 回复 上海今天多云27°C适合逛公园或咖啡馆。 最终回复上海今天多云27°C适合逛公园或咖啡馆。注意这里发生了两次工具调用第一次查天气第二次根据天气推荐活动。这说明模型在拿到第一次结果后自主决定再调一次工具。这就是 MCP 和传统 Function Calling 的区别——调用链是模型驱动的不是应用层写死的。输入exit退出Client 的__aexit__会关闭 SSE 连接和 Session生命周期进入会话关闭阶段。你可以在 Server 终端看到连接断开的日志。如果你想验证多模型切换只需改.env里的TAOTOKEN_MODEL比如换成claude-3-5-sonnet重启 Host 即可。Key 和 Base URL 都不用动这就是统一 Key 的价值。想快速对比不同模型的工具调用表现可以去模型对话页面 https://taotoken.net/models?utm_sourcetaotoken_aicg_blog_endutm_contentmodelsutm_campaignrewrite 直接试。5. 本篇常见错排查401、local proxy failed、reading choices、OAuth跑 DEMO 时最容易撞上四类报错我逐个拆。401 Unauthorized。日志里出现Error code: 401基本是 Key 问题。检查三点.env里的TAOTOKEN_API_KEY有没有多余空格Key 是不是复制时漏了前缀Base URL 是不是写成了https://taotoken.net/api/v1。正确写法是https://taotoken.net/api。如果还报 401去 https://taotoken.net/api-keys?utm_sourcetaotoken_aicg_blog_endutm_contentapi-keysutm_campaignrewrite 重新生成一个 Key 试试。local proxy failed。这个报错通常出现在你本地网络环境有额外代理设置时。MCP Client 连的是http://127.0.0.1:1234/sse这是本地回环地址不应该走任何外部通道。检查你的 shell 里有没有HTTP_PROXY或HTTPS_PROXY环境变量有的话临时 unset 掉unset HTTP_PROXY unset HTTPS_PROXY然后重启 Server 和 Client。另外确认 Server 确实在 1234 端口监听用curl http://127.0.0.1:1234/sse能看到事件流就说明正常。reading choices 报错。日志里出现KeyError: choices或reading choices说明 LLM 返回的 JSON 结构和你预期的不一样。常见原因是 Model ID 写错了TaoToken 返回了一个错误对象而不是正常的 completion。打印完整响应看看print(response.model_dump_json(indent2))确认choices字段存在。如果返回的是{error: {...}}那就是 Model ID 或 Key 的问题。去接入文档 https://taotoken.net/doc?utm_sourcetaotoken_aicg_blog_endutm_contentdocutm_campaignrewrite 核对当前支持的模型列表。OAuth 相关报错。如果你在 Claude Code 或 Cline 里配置 MCP可能会遇到OAuth token expired或authentication failed。这类工具通常有自己的认证流程和 MCP Server 本身的 SSE 连接是两回事。排查顺序先确认 TaoToken 的 Key 在工具设置里填对了Base URL 是https://taotoken.net/apiModel ID 是有效值。三件套缺一不可。如果工具提示 OAuth检查是不是把 TaoToken Key 填到了 OAuth 字段而不是 API Key 字段。还有一个隐蔽的坑MCP Server 启动后如果你改了工具代码但没重启 ServerClient 列出的还是旧工具列表。能力协商阶段拿到的工具清单是 Server 启动时注册的改代码必须重启。6. 继续深入把 DEMO 扩展成你自己的 Agent跑通这个 DEMO 后你可以做几件事让它更接近生产。第一把硬编码的天气数据换成真实 API 调用。在get_weather里发 HTTP 请求到天气服务返回真实数据。MCP Server 的价值就在这里——工具实现怎么变模型侧都不用改。第二增加 Resource。MCP 除了 Tool 还有 Resource 概念适合暴露只读数据比如“当前用户信息”“项目配置文件”。在 Server 里用mcp.resource()注册Client 用list_resources()和read_resource()访问。第三做多 Server 聚合。一个 Host 可以同时连多个 MCP Server比如天气 Server、数据库 Server、文件系统 Server。Client 侧维护多个 sessionHost 把所有工具汇总后传给模型。这样模型就能在一个对话里跨服务调用。第四加错误处理。现在工具调用失败会直接抛异常生产环境应该捕获后把错误信息作为 tool 结果回传给模型让模型决定是重试还是告知用户。如果你要长期跑这类 AgentCoding Plan 的额度模型更适合高频调用场景https://taotoken.net/coding-plan?utm_sourcetaotoken_aicg_blog_endutm_contentcoding-planutm_campaignrewrite 。接入过程中遇到协议细节问题文档里对 SSE 和 JSON-RPC 的说明比较全https://taotoken.net/doc?utm_sourcetaotoken_aicg_blog_endutm_contentdocutm_campaignrewrite 。最后提醒一个实操细节MCP 的 SSE 连接是长连接Server 和 Client 要同时运行。如果你在 Docker 里跑 Server记得把 1234 端口映射出来并且 Client 里的server_url要改成宿主机的地址不能写127.0.0.1。这个坑我在本地和容器混合部署时踩过日志里表现为连接超时但 Server 明明在跑。
用一个DEMO拆解MCP生命周期：TaoToken统一Key下的模型上下文协议实战
1. 从一次“查天气”说起MCP 生命周期到底在解决什么问题MCPModel Context Protocol模型上下文协议是什么一句话它是一套让大模型用统一方式“伸手”去拿外部数据和调用外部工具的协议。能做什么把“模型不知道的实时信息”和“模型不该自己瞎编的操作”交给外部服务处理。适合谁正在做 AI Agent、智能客服、IDE 插件、企业内部助手的开发者尤其是被 Function Calling 各家格式折磨过的人。我拿一个最小场景切入用户在聊天框里问“上海今天天气怎么样适合出门吗”。模型训练数据里没有今天的天气它必须调用一个get_weather工具拿到“上海多云27°C”再根据结果推荐活动。整个过程里MCP 要经历四个阶段初始化握手、能力协商列出有哪些工具和资源、工具调用、会话关闭。这四个阶段合起来就是 MCP 生命周期。很多人第一次接触 MCP 会把它和 Function Calling 混为一谈。区别在于Function Calling 是“应用层预先决定给模型哪些函数”而 MCP 是“模型基于上下文自主推理该调哪个工具”工具的实现细节被封装在独立的 MCP Server 里对模型透明。这意味着你新增一个工具只要符合 MCP 协议标准模型侧代码一行都不用改。这篇 DEMO 我会用 TaoToken 统一 Key 作为模型通道把 MCP Server、MCP Client、MCP Host 三段代码串起来让你能亲手跑通一次完整的工具调用并在日志里看到生命周期每个阶段的真实输出。TaoToken 在这里的作用是一个 Key 就能切换不同模型省去为每个模型单独配 Key 的麻烦特别适合做多模型接入验证。2. TaoToken 前置准备统一 Key 与 API 通道配置在写 MCP 代码之前先把模型通道打通。TaoToken 的定位是统一 API 通道你注册后拿到一个 Key就能通过兼容 OpenAI 格式的接口调用多种模型。官网入口是 https://taotoken.net/?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_content API 基址是 https://taotoken.net/api 。第一步去控制台创建 API Key。打开 https://taotoken.net/console?utm_sourcetaotoken_aicg_blog_endutm_contentconsoleutm_campaignrewrite 在 API Keys 页面点新建复制生成的 Key。这个 Key 后面会同时用在 MCP Host 的 LLM 调用里。第二步确认你要用的模型 ID。在模型对话页面 https://taotoken.net/models?utm_sourcetaotoken_aicg_blog_endutm_contentmodelsutm_campaignrewrite 可以看到当前支持的模型列表选一个你熟悉的比如gpt-4o-mini或claude-3-5-sonnet。记下这个 Model ID配置里要用。第三步把 Key 和 Base URL 写进环境变量避免硬编码进代码。在项目根目录建一个.env文件TAOTOKEN_API_KEYsk-你的Key TAOTOKEN_BASE_URLhttps://taotoken.net/api TAOTOKEN_MODELgpt-4o-mini然后在 Python 里用python-dotenv读取。如果你不想装额外依赖也可以直接在 shell 里 exportexport TAOTOKEN_API_KEYsk-你的Key export TAOTOKEN_BASE_URLhttps://taotoken.net/api export TAOTOKEN_MODELgpt-4o-mini这里有个容易踩的坑Base URL 末尾不要带/v1TaoToken 的兼容层会自动补全路径。如果你手动拼成https://taotoken.net/api/v1/chat/completions反而可能 404。正确的请求地址是https://taotoken.net/api/chat/completions。另外MCP Server 本身不需要 TaoToken Key它只负责提供工具。Key 只用在 MCP Host 调用 LLM 的那一步。这个分工要理清楚否则你会以为 Server 也要配 Key。如果你打算长期跑编码类 Agent可以了解下 Coding Planhttps://taotoken.net/coding-plan?utm_sourcetaotoken_aicg_blog_endutm_contentcoding-planutm_campaignrewrite 它针对高频编码场景做了额度优化。接入文档在 https://taotoken.net/doc?utm_sourcetaotoken_aicg_blog_endutm_contentdocutm_campaignrewrite 遇到参数问题可以先查这里。3. 可复制配置MCP Server、Client、Host 三段代码这一节是全文核心我把三段代码完整贴出来你复制就能跑。先装依赖pip install mcp openai python-dotenv注意这里用openai库而不是requests因为 TaoToken 兼容 OpenAI 格式用官方 SDK 更省事也方便你以后换模型。3.1 MCP Server注册两个工具新建mcp_server_demo.pyfrom mcp.server.fastmcp import FastMCP import asyncio mcp FastMCP(nameweather-demo, host0.0.0.0, port1234) mcp.tool(nameget_weather, description获取指定城市的天气信息) async def get_weather(city: str) - str: weather_data { 北京: 北京晴25°C, 上海: 上海多云27°C, 广州: 广州小雨30°C } return weather_data.get(city, f{city}天气信息未知) mcp.tool(namesuggest_activity, description根据天气描述推荐适合的活动) async def suggest_activity(condition: str) - str: if 晴 in condition: return 天气晴朗推荐你去户外散步或运动。 elif 多云 in condition: return 多云天气适合逛公园或咖啡馆。 elif 雨 in condition: return 下雨了建议你在家阅读或看电影。 else: return 建议进行室内活动。 async def main(): print(启动 MCP Server: http://127.0.0.1:1234) await mcp.run_sse_async() if __name__ __main__: asyncio.run(main())这段代码用FastMCP装饰器注册了两个工具。run_sse_async()会启动一个 SSE 服务监听 1234 端口。启动后你会看到启动 MCP Server: http://127.0.0.1:1234。3.2 MCP Client连接 Server 并列出能力新建mcp_client_demo.pyimport asyncio from mcp.client.session import ClientSession from mcp.client.sse import sse_client class WeatherMCPClient: def __init__(self, server_urlhttp://127.0.0.1:1234/sse): self.server_url server_url self._sse_context None self._session None async def __aenter__(self): self._sse_context sse_client(self.server_url) self.read, self.write await self._sse_context.__aenter__() self._session ClientSession(self.read, self.write) await self._session.__aenter__() await self._session.initialize() return self async def __aexit__(self, exc_type, exc_val, exc_tb): if self._session: await self._session.__aexit__(exc_type, exc_val, exc_tb) if self._sse_context: await self._sse_context.__aexit__(exc_type, exc_val, exc_tb) async def list_tools(self): return await self._session.list_tools() async def list_resources(self): return await self._session.list_resources() async def call_tool(self, name, arguments): return await self._session.call_tool(name, arguments) async def main(): async with WeatherMCPClient() as client: print(成功连接 MCP Server) tools await client.list_tools() print(\n可用工具:) print(tools) resources await client.list_resources() print(\n可用资源:) print(resources) print(\n调用 get_weather 工具(city上海)...) result await client.call_tool(get_weather, {city: 上海}) print(\n工具返回:) for item in result.content: print( -, item.text) if __name__ __main__: asyncio.run(main())__aenter__里做了三件事建立 SSE 通道、创建 ClientSession、调用initialize()完成握手。这就是生命周期的初始化阶段。list_tools()是能力协商阶段call_tool()是工具调用阶段__aexit__是会话关闭阶段。3.3 MCP Host串起 LLM 和工具新建mcp_host_demo.pyimport asyncio import json import re import os from openai import OpenAI from dotenv import load_dotenv from mcp_client_demo import WeatherMCPClient load_dotenv() client_llm OpenAI( api_keyos.getenv(TAOTOKEN_API_KEY), base_urlos.getenv(TAOTOKEN_BASE_URL) ) MODEL_ID os.getenv(TAOTOKEN_MODEL, gpt-4o-mini) def extract_json_from_reply(reply: str): if isinstance(reply, dict): return reply if isinstance(reply, str): reply re.sub(r^(?:json)?|$, , reply.strip(), flagsre.IGNORECASE).strip() for _ in range(3): try: parsed json.loads(reply) if isinstance(parsed, dict): return parsed else: reply parsed except Exception: break return reply async def main(): client WeatherMCPClient() await client.__aenter__() tools await client.list_tools() resources await client.list_resources() tool_names [t.name for t in tools.tools] tool_descriptions \n.join(f- {t.name}: {t.description} for t in tools.tools) resource_descriptions \n.join(f- {r.uri} for r in resources.resources) while True: user_input input(\n请输入你的问题输入 exit 退出\n ) if user_input.lower() in (exit, 退出): break system_prompt ( 你是一个智能助手拥有以下工具和资源可以调用\n\n f工具列表\n{tool_descriptions or 无}\n\n f资源列表\n{resource_descriptions or 无}\n\n 请优先调用可用的 Tool 或 Resource而不是 llm 内部生成。 仅根据上下文调用工具不传入不需要的参数进行调用\n 如果需要请以 JSON 返回 tool_calls格式如下\n {tool_calls: [{name: get_weather, arguments: {city: 北京}}]}\n 如无需调用工具返回{\tool_calls\: null} ) messages [ {role: system, content: system_prompt}, {role: user, content: user_input} ] final_reply while True: response client_llm.chat.completions.create( modelMODEL_ID, messagesmessages ) reply response.choices[0].message.content print(f\nLLM 回复\n{reply}) parsed extract_json_from_reply(reply) if isinstance(parsed, str): final_reply parsed break tool_calls parsed.get(tool_calls) if not tool_calls: final_reply parsed.get(content, ) break for tool_call in tool_calls: tool_name tool_call[name] arguments tool_call[arguments] if tool_name not in tool_names: raise ValueError(f工具 {tool_name} 未注册) print(f调用工具 {tool_name} 参数: {arguments}) result await client.call_tool(tool_name, arguments) tool_output result.content[0].text print(f工具 {tool_name} 返回{tool_output}) messages.append({ role: tool, name: tool_name, content: tool_output }) print(f\n最终回复{final_reply}) await client.__aexit__(None, None, None) if __name__ __main__: asyncio.run(main())这里的关键点client_llm用的是 TaoToken 的 Base URL 和 KeyModel ID 从环境变量读。messages里追加role: tool的消息就是把工具结果回传给模型。整个循环直到模型返回纯文本才结束。如果你用 Claude Code 或 Cline 这类工具配置方式类似需要填三件套Base URL 填https://taotoken.net/apiKey 填你的 TaoToken KeyModel ID 填你选的模型。Claude Code 的接入文档在 https://taotoken.net/claude-code?utm_sourcetaotoken_aicg_blog_endutm_contentclaude-codeutm_campaignrewrite 。4. 验证请求生命周期各阶段日志与成功结果先启动 Serverpython mcp_server_demo.py看到启动 MCP Server: http://127.0.0.1:1234就说明初始化成功。这个阶段对应生命周期的“初始化握手”Server 在 1234 端口等待 SSE 连接。另开一个终端先单独测 Clientpython mcp_client_demo.py你应该看到成功连接 MCP Server 可用工具: metaNone nextCursorNone tools[Tool(nameget_weather, description获取指定城市的天气信息, inputSchema{...}), Tool(namesuggest_activity, ...)] 可用资源: metaNone nextCursorNone resources[] 调用 get_weather 工具(city上海)... 工具返回: - 上海多云27°C这段日志覆盖了三个生命周期阶段成功连接是初始化可用工具是能力协商工具返回是工具调用。可用资源为空是因为我们没注册 resource不影响 DEMO。现在跑完整的 Hostpython mcp_host_demo.py输入“上海今天天气怎么样适合出门吗”你会看到类似输出LLM 回复 {tool_calls: [{name: get_weather, arguments: {city: 上海}}]} 调用工具 get_weather 参数: {city: 上海} 工具 get_weather 返回上海多云27°C LLM 回复 {tool_calls: [{name: suggest_activity, arguments: {condition: 多云}}]} 调用工具 suggest_activity 参数: {condition: 多云} 工具 suggest_activity 返回多云天气适合逛公园或咖啡馆。 LLM 回复 上海今天多云27°C适合逛公园或咖啡馆。 最终回复上海今天多云27°C适合逛公园或咖啡馆。注意这里发生了两次工具调用第一次查天气第二次根据天气推荐活动。这说明模型在拿到第一次结果后自主决定再调一次工具。这就是 MCP 和传统 Function Calling 的区别——调用链是模型驱动的不是应用层写死的。输入exit退出Client 的__aexit__会关闭 SSE 连接和 Session生命周期进入会话关闭阶段。你可以在 Server 终端看到连接断开的日志。如果你想验证多模型切换只需改.env里的TAOTOKEN_MODEL比如换成claude-3-5-sonnet重启 Host 即可。Key 和 Base URL 都不用动这就是统一 Key 的价值。想快速对比不同模型的工具调用表现可以去模型对话页面 https://taotoken.net/models?utm_sourcetaotoken_aicg_blog_endutm_contentmodelsutm_campaignrewrite 直接试。5. 本篇常见错排查401、local proxy failed、reading choices、OAuth跑 DEMO 时最容易撞上四类报错我逐个拆。401 Unauthorized。日志里出现Error code: 401基本是 Key 问题。检查三点.env里的TAOTOKEN_API_KEY有没有多余空格Key 是不是复制时漏了前缀Base URL 是不是写成了https://taotoken.net/api/v1。正确写法是https://taotoken.net/api。如果还报 401去 https://taotoken.net/api-keys?utm_sourcetaotoken_aicg_blog_endutm_contentapi-keysutm_campaignrewrite 重新生成一个 Key 试试。local proxy failed。这个报错通常出现在你本地网络环境有额外代理设置时。MCP Client 连的是http://127.0.0.1:1234/sse这是本地回环地址不应该走任何外部通道。检查你的 shell 里有没有HTTP_PROXY或HTTPS_PROXY环境变量有的话临时 unset 掉unset HTTP_PROXY unset HTTPS_PROXY然后重启 Server 和 Client。另外确认 Server 确实在 1234 端口监听用curl http://127.0.0.1:1234/sse能看到事件流就说明正常。reading choices 报错。日志里出现KeyError: choices或reading choices说明 LLM 返回的 JSON 结构和你预期的不一样。常见原因是 Model ID 写错了TaoToken 返回了一个错误对象而不是正常的 completion。打印完整响应看看print(response.model_dump_json(indent2))确认choices字段存在。如果返回的是{error: {...}}那就是 Model ID 或 Key 的问题。去接入文档 https://taotoken.net/doc?utm_sourcetaotoken_aicg_blog_endutm_contentdocutm_campaignrewrite 核对当前支持的模型列表。OAuth 相关报错。如果你在 Claude Code 或 Cline 里配置 MCP可能会遇到OAuth token expired或authentication failed。这类工具通常有自己的认证流程和 MCP Server 本身的 SSE 连接是两回事。排查顺序先确认 TaoToken 的 Key 在工具设置里填对了Base URL 是https://taotoken.net/apiModel ID 是有效值。三件套缺一不可。如果工具提示 OAuth检查是不是把 TaoToken Key 填到了 OAuth 字段而不是 API Key 字段。还有一个隐蔽的坑MCP Server 启动后如果你改了工具代码但没重启 ServerClient 列出的还是旧工具列表。能力协商阶段拿到的工具清单是 Server 启动时注册的改代码必须重启。6. 继续深入把 DEMO 扩展成你自己的 Agent跑通这个 DEMO 后你可以做几件事让它更接近生产。第一把硬编码的天气数据换成真实 API 调用。在get_weather里发 HTTP 请求到天气服务返回真实数据。MCP Server 的价值就在这里——工具实现怎么变模型侧都不用改。第二增加 Resource。MCP 除了 Tool 还有 Resource 概念适合暴露只读数据比如“当前用户信息”“项目配置文件”。在 Server 里用mcp.resource()注册Client 用list_resources()和read_resource()访问。第三做多 Server 聚合。一个 Host 可以同时连多个 MCP Server比如天气 Server、数据库 Server、文件系统 Server。Client 侧维护多个 sessionHost 把所有工具汇总后传给模型。这样模型就能在一个对话里跨服务调用。第四加错误处理。现在工具调用失败会直接抛异常生产环境应该捕获后把错误信息作为 tool 结果回传给模型让模型决定是重试还是告知用户。如果你要长期跑这类 AgentCoding Plan 的额度模型更适合高频调用场景https://taotoken.net/coding-plan?utm_sourcetaotoken_aicg_blog_endutm_contentcoding-planutm_campaignrewrite 。接入过程中遇到协议细节问题文档里对 SSE 和 JSON-RPC 的说明比较全https://taotoken.net/doc?utm_sourcetaotoken_aicg_blog_endutm_contentdocutm_campaignrewrite 。最后提醒一个实操细节MCP 的 SSE 连接是长连接Server 和 Client 要同时运行。如果你在 Docker 里跑 Server记得把 1234 端口映射出来并且 Client 里的server_url要改成宿主机的地址不能写127.0.0.1。这个坑我在本地和容器混合部署时踩过日志里表现为连接超时但 Server 明明在跑。
