1. 从一次“查天气”说起：MCP 生命周期到底在解决什么问题
MCP（Model 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_source=taotoken_aicg_blog_end&utm_medium=csdn&utm_campaign=rewrite&utm_content= ，API 基址是 https://taotoken.net/api 。
第一步，去控制台创建 API Key。打开 https://taotoken.net/console?utm_source=taotoken_aicg_blog_end&utm_content=console&utm_campaign=rewrite ，在 API Keys 页面点新建，复制生成的 Key。这个 Key 后面会同时用在 MCP Host 的 LLM 调用里。
第二步，确认你要用的模型 ID。在模型对话页面 https://taotoken.net/models?utm_source=taotoken_aicg_blog_end&utm_content=models&utm_campaign=rewrite 可以看到当前支持的模型列表，选一个你熟悉的，比如gpt-4o-mini或claude-3-5-sonnet。记下这个 Model ID，配置里要用。
第三步，把 Key 和 Base URL 写进环境变量，避免硬编码进代码。在项目根目录建一个.env文件：
TAOTOKEN_API_KEY=sk-你的Key TAOTOKEN_BASE_URL=https://taotoken.net/api TAOTOKEN_MODEL=gpt-4o-mini然后在 Python 里用python-dotenv读取。如果你不想装额外依赖，也可以直接在 shell 里 export：
export TAOTOKEN_API_KEY="sk-你的Key" export TAOTOKEN_BASE_URL="https://taotoken.net/api" export TAOTOKEN_MODEL="gpt-4o-mini"这里有个容易踩的坑：Base URL 末尾不要带/v1，TaoToken 的兼容层会自动补全路径。如果你手动拼成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 Plan：https://taotoken.net/coding-plan?utm_source=taotoken_aicg_blog_end&utm_content=coding-plan&utm_campaign=rewrite ，它针对高频编码场景做了额度优化。接入文档在 https://taotoken.net/doc?utm_source=taotoken_aicg_blog_end&utm_content=doc&utm_campaign=rewrite ，遇到参数问题可以先查这里。
3. 可复制配置：MCP Server、Client、Host 三段代码
这一节是全文核心，我把三段代码完整贴出来，你复制就能跑。先装依赖：
pip install mcp openai python-dotenv注意这里用openai库而不是requests，因为 TaoToken 兼容 OpenAI 格式，用官方 SDK 更省事，也方便你以后换模型。
3.1 MCP Server：注册两个工具
新建mcp_server_demo.py：
from mcp.server.fastmcp import FastMCP import asyncio mcp = FastMCP(name="weather-demo", host="0.0.0.0", port=1234) @mcp.tool(name="get_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(name="suggest_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.py：
import asyncio from mcp.client.session import ClientSession from mcp.client.sse import sse_client class WeatherMCPClient: def __init__(self, server_url="http://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.py：
import 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_key=os.getenv("TAOTOKEN_API_KEY"), base_url=os.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(), flags=re.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( model=MODEL_ID, messages=messages ) 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 和 Key，Model ID 从环境变量读。messages里追加role: tool的消息，就是把工具结果回传给模型。整个循环直到模型返回纯文本才结束。
如果你用 Claude Code 或 Cline 这类工具，配置方式类似，需要填三件套：Base URL 填https://taotoken.net/api，Key 填你的 TaoToken Key，Model ID 填你选的模型。Claude Code 的接入文档在 https://taotoken.net/claude-code?utm_source=taotoken_aicg_blog_end&utm_content=claude-code&utm_campaign=rewrite 。
4. 验证请求：生命周期各阶段日志与成功结果
先启动 Server：
python mcp_server_demo.py看到启动 MCP Server: http://127.0.0.1:1234就说明初始化成功。这个阶段对应生命周期的“初始化握手”，Server 在 1234 端口等待 SSE 连接。
另开一个终端，先单独测 Client：
python mcp_client_demo.py你应该看到：
成功连接 MCP Server 可用工具: meta=None nextCursor=None tools=[Tool(name='get_weather', description='获取指定城市的天气信息', inputSchema={...}), Tool(name='suggest_activity', ...)] 可用资源: meta=None nextCursor=None resources=[] 调用 get_weather 工具(city=上海)... 工具返回: - 上海：多云，27°C这段日志覆盖了三个生命周期阶段：成功连接是初始化，可用工具是能力协商，工具返回是工具调用。可用资源为空是因为我们没注册 resource，不影响 DEMO。
现在跑完整的 Host：
python 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_source=taotoken_aicg_blog_end&utm_content=models&utm_campaign=rewrite 直接试。
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_source=taotoken_aicg_blog_end&utm_content=api-keys&utm_campaign=rewrite 重新生成一个 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(indent=2))确认choices字段存在。如果返回的是{"error": {...}}，那就是 Model ID 或 Key 的问题。去接入文档 https://taotoken.net/doc?utm_source=taotoken_aicg_blog_end&utm_content=doc&utm_campaign=rewrite 核对当前支持的模型列表。
OAuth 相关报错。如果你在 Claude Code 或 Cline 里配置 MCP，可能会遇到OAuth token expired或authentication failed。这类工具通常有自己的认证流程，和 MCP Server 本身的 SSE 连接是两回事。排查顺序：先确认 TaoToken 的 Key 在工具设置里填对了，Base URL 是https://taotoken.net/api，Model ID 是有效值。三件套缺一不可。如果工具提示 OAuth，检查是不是把 TaoToken Key 填到了 OAuth 字段而不是 API Key 字段。
还有一个隐蔽的坑：MCP Server 启动后，如果你改了工具代码但没重启 Server，Client 列出的还是旧工具列表。能力协商阶段拿到的工具清单是 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 侧维护多个 session，Host 把所有工具汇总后传给模型。这样模型就能在一个对话里跨服务调用。
第四，加错误处理。现在工具调用失败会直接抛异常，生产环境应该捕获后把错误信息作为 tool 结果回传给模型，让模型决定是重试还是告知用户。
如果你要长期跑这类 Agent，Coding Plan 的额度模型更适合高频调用场景：https://taotoken.net/coding-plan?utm_source=taotoken_aicg_blog_end&utm_content=coding-plan&utm_campaign=rewrite 。接入过程中遇到协议细节问题，文档里对 SSE 和 JSON-RPC 的说明比较全：https://taotoken.net/doc?utm_source=taotoken_aicg_blog_end&utm_content=doc&utm_campaign=rewrite 。
最后提醒一个实操细节：MCP 的 SSE 连接是长连接，Server 和 Client 要同时运行。如果你在 Docker 里跑 Server，记得把 1234 端口映射出来，并且 Client 里的server_url要改成宿主机的地址，不能写127.0.0.1。这个坑我在本地和容器混合部署时踩过，日志里表现为连接超时，但 Server 明明在跑。
