1. 凌晨三点的告警，和那个不想起床的你
服务器 CPU 突然飙到 98%，手机在床头柜上震个不停。你爬起来打开电脑，SSH 连上去，top看一眼进程，tail -f翻日志，再去内网 Wiki 搜一下这个报错以前有没有出现过——这套动作你做过不下五十次，闭着眼都能敲完。问题是，它消耗的不是技术能力，而是你的睡眠和耐心。
大模型现在写代码、读日志、做推理的能力已经足够强，为什么还要人肉去执行这些标准化步骤？我想要的是一段 Python 脚本，收到告警后自动拉取服务器状态、读取错误日志、结合本地运维手册，最后把一份“诊断报告”推给我。我只需要看一眼结论，决定要不要回滚。
这就是本文要手搓的东西：一个基于 Python + MCP + LLM 的智能服务器运维助手。MCP（Model Context Protocol）负责把本地服务器操作封装成标准工具接口，LLM 负责推理和决策，Python 负责把两者粘起来。适合谁？适合有一定 Python 基础、手头管着几台 Linux 服务器、想用 AI 把重复运维动作自动化掉的开发者。
但这里有个现实问题：当你真正开始写 Agent 的时候，会发现模型调用的 Key 管理很烦。今天用 DeepSeek 试推理，明天想换 Claude 做长日志分析，后天团队里另一个人用 GPT 跑同样的流程——每个模型一套 Key、一套 Base URL、一套计费，散落在各个.env文件里。切换一次要改代码、改环境变量、重启服务。我试过在三个项目里维护四套 Key，最后自己都记不清哪个 Key 对应哪个模型。
所以这篇教程的重点不只是“怎么搭一个 MCP 运维助手”，而是“怎么把本地 MCP 服务的 endpoint 和鉴权统一改到 TaoToken 通道”，让多模型切换变成改一个 Model ID 的事。下面从环境准备开始，一步步跑通从本地工具调用到模型响应的完整闭环。
2. 把 MCP 服务接到 TaoToken：环境准备与依赖安装
在写代码之前，先把思路理清楚。整个系统分三层：工具层是一个 MCP Server，用 Python 写，暴露get_server_status和read_last_error_logs两个工具；大脑层是一个 Python Agent 主程序，它作为 MCP Client 连接 Server，同时调用 LLM 做推理；通道层就是 TaoToken，统一管理模型访问。
为什么要把 MCP 服务的模型调用改到 TaoToken？因为 MCP 协议本身只规定了“工具怎么暴露、怎么调用”，它不关心你背后用哪个模型。但你的 Agent 主程序在调用 LLM 时，需要填base_url和api_key。如果每个模型都单独配，代码里就会散落一堆 if-else。TaoToken 提供统一的 API 入口，你只需要把base_url指向https://taotoken.net/api，然后用同一个 Key 就能访问不同模型。切换模型时只改model参数，不动鉴权逻辑。
先装依赖。Python 版本建议 3.10 以上，MCP 的 Python SDK 对异步支持要求较高。
pip install mcp openai psutilmcp是官方 SDK，openai用来调用兼容 OpenAI 格式的接口（TaoToken 的 API 就是这种格式），psutil用来读 CPU 和内存。
接下来配置环境变量。我习惯用一个.env文件管理，但为了演示清晰，这里直接写在 shell 里。你需要去 TaoToken 控制台创建一个 API Key，然后设置：
export TAOTOKEN_API_KEY="sk-你的Key" export TAOTOKEN_BASE_URL="https://taotoken.net/api"注意，base_url后面不要加/v1，OpenAI SDK 会自动拼接。如果你用的是其他兼容库，确认一下它的拼接规则。我踩过的坑是：有些教程写https://taotoken.net/api/v1，结果请求变成/api/v1/v1/chat/completions，直接 404。
MCP Server 的配置也需要一个环境变量模板。如果你打算把 Server 和 Agent 分开部署（比如 Server 跑在目标服务器上，Agent 跑在本地），可以用一个mcp_config.json来管理连接参数：
{ "mcpServers": { "ops-watchdog": { "command": "python", "args": ["/opt/ops/server_mcp.py"], "env": { "TAOTOKEN_API_KEY": "sk-你的Key", "TAOTOKEN_BASE_URL": "https://taotoken.net/api" } } } }这个 JSON 片段可以直接被支持 MCP 的客户端读取。如果你用 Claude Code 或 Cline，它们的配置文件格式类似，把command和args换成你的实际路径即可。这里的关键是：MCP Server 本身不直接调用 LLM，它只负责执行工具；LLM 调用发生在 Agent 主程序里。所以TAOTOKEN_API_KEY其实是给 Agent 用的，放在 Server 的 env 里只是为了统一管理，实际读取时要注意作用域。
如果你用 Codex 的auth.json做鉴权，格式是这样的：
{ "api_key": "sk-你的Key", "base_url": "https://taotoken.net/api" }但 Codex 的auth.json通常放在~/.codex/目录下，而我们的 Agent 是独立 Python 程序，直接读环境变量更简单。所以下面的代码里，我用os.getenv读取，不依赖外部配置文件。
依赖装好、Key 配好之后，先写 MCP Server。新建server_mcp.py：
import sys import psutil import subprocess from mcp.server.fastmcp import FastMCP mcp = FastMCP("OpsWatchdog") @mcp.tool() def get_server_status() -> str: """获取服务器当前的 CPU 和内存使用率。返回格式化字符串。""" cpu_percent = psutil.cpu_percent(interval=1) memory = psutil.virtual_memory() return f"CPU使用率: {cpu_percent}%\n内存使用率: {memory.percent}%" @mcp.tool() def read_last_error_logs(lines: int = 50) -> str: """读取应用日志文件的最后 N 行，用于分析报错原因。""" log_path = "/var/log/syslog" try: result = subprocess.check_output(['tail', '-n', str(lines), log_path]) return result.decode('utf-8') except Exception as e: return f"读取日志失败: {str(e)}" if __name__ == "__main__": print("MCP Server Running...", file=sys.stderr) mcp.run()这段代码里，FastMCP是 SDK 提供的快捷类，@mcp.tool()装饰器把普通 Python 函数注册成 MCP 工具。mcp.run()默认用 stdio 传输，也就是通过标准输入输出和 Client 通信。注意print要写到stderr，因为stdout被 MCP 协议占用了，写错地方会导致协议解析失败。
Server 写完后，先单独测一下它能不能启动：
python server_mcp.py如果看到MCP Server Running...并且没有报错，说明 Server 端没问题。按Ctrl+C退出，接下来写 Agent 主程序。
3. 可复制配置：Agent 主程序与 TaoToken 接入
Agent 主程序的核心逻辑是：启动 MCP Server 子进程 → 建立 ClientSession → 获取工具列表 → 把工具描述注入 System Prompt → 调用 LLM → 解析 LLM 的工具调用意图 → 执行工具 → 把结果喂回 LLM → 输出最终诊断。
新建agent_main.py：
import asyncio import os import json from mcp import ClientSession, StdioServerParameters from mcp.client.stdio import stdio_client from openai import OpenAI client = OpenAI( api_key=os.getenv("TAOTOKEN_API_KEY"), base_url=os.getenv("TAOTOKEN_BASE_URL", "https://taotoken.net/api") ) async def run_agent(): server_params = StdioServerParameters( command="python", args=["server_mcp.py"], ) async with stdio_client(server_params) as (read, write): async with ClientSession(read, write) as session: await session.initialize() tools = await session.list_tools() tool_descriptions = "\n".join( [f"- {t.name}: {t.description}" for t in tools.tools] ) system_prompt = f"""你是一个资深 SRE 运维专家。你可以使用以下工具排查服务器问题： {tool_descriptions} 当收到报警时，请先调用工具获取信息，然后分析原因。 输出格式要求：先给出诊断结论，再给出修复建议。""" user_query = "服务器 CPU 突然飙升，请排查原因并给出建议。" print(f"用户指令: {user_query}") response = client.chat.completions.create( model="deepseek-chat", messages=[ {"role": "system", "content": system_prompt}, {"role": "user", "content": user_query} ] ) llm_thought = response.choices[0].message.content print(f"Agent 思考: {llm_thought}") if "get_server_status" in llm_thought: print(">> Agent 决定调用工具: get_server_status") tool_result = await session.call_tool("get_server_status") result_text = tool_result.content[0].text print(f">> 工具返回结果: {result_text}") final_response = client.chat.completions.create( model="deepseek-chat", messages=[ {"role": "system", "content": system_prompt}, {"role": "user", "content": user_query}, {"role": "assistant", "content": llm_thought}, {"role": "user", "content": f"工具执行结果: {result_text}"} ] ) print(f"Agent 最终诊断: {final_response.choices[0].message.content}") if __name__ == "__main__": asyncio.run(run_agent())这段代码里，model="deepseek-chat"是你要根据实际需求改的。TaoToken 支持多种模型，你可以在控制台看到可用的 Model ID 列表。想换成 Claude 做长日志分析，就把model改成对应的 ID，base_url和api_key不用动。这就是统一通道的价值：鉴权逻辑只写一次，模型切换只改一个字符串。
如果你用 Cline 或 Claude Code 的 MCP 配置，把上面的server_params换成对应的 JSON 配置即可。Cline 的 MCP 设置里，command填python，args填["server_mcp.py"]，env里加上TAOTOKEN_API_KEY和TAOTOKEN_BASE_URL。Claude Code 的配置类似，但它的 MCP 配置文件通常在~/.claude/claude_desktop_config.json，格式和前面的mcp_config.json一致。
这里有个细节要注意：MCP Server 的env和 Agent 的env是两套。Server 不需要调 LLM，所以它的env里其实不需要TAOTOKEN_API_KEY。但如果你把 Server 和 Agent 写在同一个进程里（比如用FastMCP的run方法直接启动），那 Key 就在同一个环境里。我建议分开写，职责清晰，排障也容易。
配置写完后，先别急着跑。检查一下server_mcp.py的路径是否正确，args里的文件名要和实际一致。如果你把两个文件放在不同目录，args要写绝对路径，否则stdio_client找不到文件会报FileNotFoundError。
4. 验证请求：一次真实的运维指令端到端跑通
现在跑一次完整的流程。确保你在agent_main.py和server_mcp.py所在的目录下，并且环境变量已经设置好：
export TAOTOKEN_API_KEY="sk-你的Key" export TAOTOKEN_BASE_URL="https://taotoken.net/api" python agent_main.py如果一切正常，你会看到类似下面的输出：
用户指令: 服务器 CPU 突然飙升，请排查原因并给出建议。 Agent 思考: 我需要先查看当前的服务器状态，确认 CPU 使用率。调用 get_server_status。 >> Agent 决定调用工具: get_server_status >> 工具返回结果: CPU使用率: 98.5% 内存使用率: 72.3% Agent 最终诊断: [高危警告] 检测到服务器 CPU 使用率高达 98.5%。 结合知识库判断，可能是计算密集型任务卡死。 建议立即 SSH 登录使用 top 命令定位具体 PID，并考虑执行 kill 操作。这个输出说明闭环跑通了：Agent 收到自然语言指令 → LLM 决定调用工具 → MCP Client 通过 stdio 调用 Server → Server 执行psutil读取真实 CPU 数据 → 结果返回给 LLM → LLM 生成诊断报告。
但这里有个问题：上面的代码只演示了一次工具调用。真实的运维场景可能需要多轮：先查 CPU，发现高，再查日志，发现某个进程报错，再查知识库，给出修复建议。要实现多轮 ReAct 循环，需要解析 LLM 返回的工具调用意图，而不是简单用if "get_server_status" in llm_thought判断。
更健壮的做法是让 LLM 输出 JSON 格式的工具调用请求。修改 System Prompt：
system_prompt = f"""你是一个资深 SRE 运维专家。你可以使用以下工具排查服务器问题： {tool_descriptions} 当收到报警时，请先调用工具获取信息，然后分析原因。 如果你需要调用工具，请输出 JSON 格式：{{"tool": "工具名", "args": {{"参数名": "参数值"}}}} 如果不需要调用工具，直接输出诊断结论。"""然后在 Agent 里解析 JSON：
import json def parse_tool_call(text): try: start = text.find("{") end = text.rfind("}") + 1 if start != -1 and end != 0: return json.loads(text[start:end]) except json.JSONDecodeError: pass return None这样就能支持多轮循环。你可以把run_agent改成一个while循环，每次 LLM 返回后检查是否有工具调用请求，有就执行，把结果追加到消息历史里，继续下一轮，直到 LLM 输出最终诊断。
验证的时候，你可以故意制造一个场景：把read_last_error_logs的log_path改成一个不存在的文件，看 Agent 怎么处理错误。正常情况下，工具会返回读取日志失败: ...，LLM 应该能识别这个错误并给出“日志路径可能配置错误”的建议。这能测试整个链路的容错能力。
如果你在验证时遇到401错误，说明TAOTOKEN_API_KEY没设置对，或者 Key 已经失效。去 TaoToken 控制台重新生成一个，然后export一下。如果遇到local proxy failed，检查你的网络环境是否能正常访问https://taotoken.net/api。如果遇到reading choices相关的报错，通常是model参数填错了，去控制台确认一下可用的 Model ID。
5. 本篇常见错排查：401、local proxy failed、reading choices、OAuth
排障这部分我按真实遇到的报错来写，每个都给出定位方法和修复步骤。
401 Unauthorized
这是最常见的。报错信息通常是：
openai.AuthenticationError: Error code: 401 - {'error': {'message': 'Invalid API key'}}原因有三个：Key 没设置、Key 写错了、Key 被禁用了。先检查环境变量：
echo $TAOTOKEN_API_KEY如果输出为空，说明没export成功。如果你是在.env文件里写的，确认代码里有没有load_dotenv()。如果 Key 看起来正常，去 TaoToken 控制台确认一下这个 Key 的状态，是不是被删了或者过期了。
local proxy failed
这个报错通常出现在网络层：
openai.APIConnectionError: Connection error: local proxy failed意思是 SDK 尝试通过本地代理访问 API，但代理不可用。检查你的环境变量里有没有HTTP_PROXY或HTTPS_PROXY：
env | grep -i proxy如果有，而且你不需要代理，直接unset掉：
unset HTTP_PROXY unset HTTPS_PROXY然后重新跑。如果你确实需要代理才能访问外网，确认代理服务是否正常运行。但注意，TaoToken 的 API 地址是https://taotoken.net/api，确保你的网络环境能正常解析和访问这个域名。
reading choices 报错
这个报错通常长这样：
KeyError: 'choices'或者：
IndexError: list index out of range原因是response.choices[0]取不到值。可能是model参数填错了，API 返回了一个错误对象而不是正常的 completion 对象。打印完整的response看看：
print(response)如果看到error字段，里面会写明具体原因。常见的是 Model ID 不存在，比如你写了deepseek但实际应该是deepseek-chat。去 TaoToken 控制台的模型列表里核对一下。
OAuth 相关报错
如果你用 Claude Code 或 Cline 的 OAuth 流程接入，可能会遇到：
OAuth token expired或者：
Invalid OAuth callback这类问题通常出现在客户端工具的鉴权环节，不是 MCP Server 本身的问题。检查你的客户端工具版本，确认 OAuth 配置里的client_id和redirect_uri是否正确。如果你只是用 Python 脚本调 API，不走 OAuth，可以忽略这类报错。
MCP 连接失败
如果 Agent 启动时报：
FileNotFoundError: [Errno 2] No such file or directory: 'server_mcp.py'说明StdioServerParameters里的args路径不对。改成绝对路径：
args=["/full/path/to/server_mcp.py"]如果报：
mcp.server.fastmcp.exceptions.MCPError: Tool not found说明 LLM 请求的工具名和 Server 注册的不一致。检查@mcp.tool()装饰的函数名，确保和 System Prompt 里描述的一致。
排障的核心思路是：先确认 Key 和 Base URL 正确，再确认模型 ID 存在，最后确认 MCP 工具注册和调用匹配。这三步过了，基本不会有大问题。
6. 从本地工具到模型响应：把闭环用起来
跑通一次验证之后，你可以把这个 Agent 挂到定时任务或 Webhook 上。比如用cron每五分钟检查一次 CPU，超过阈值就触发 Agent 诊断，把结果推到钉钉或飞书。MCP Server 那边只需要加一个新的@mcp.tool()函数，比如restart_service(service_name)，Agent 就能学会新技能。
如果你想让 Agent 支持更多模型，比如用 Claude 做长日志分析、用 GPT 做代码修复建议，只需要在 TaoToken 控制台确认对应的 Model ID，然后在client.chat.completions.create里改model参数。base_url和api_key始终不变。这就是统一通道的意义：把鉴权收敛到一处，把模型选择变成配置项。
如果你还没有 TaoToken 的 Key，可以去控制台创建一个，然后从 API Keys 页面复制。接入文档里有不同语言的调用示例，Python 的示例和本文的代码结构一致。想先试试模型对话效果，可以直接在模型对话页面发一条运维相关的指令，看看不同模型的响应差异。如果你打算长期跑编码或 Agent 任务，Coding Plan 的额度更划算，适合把这类自动化脚本部署到生产环境。
最后留一个实用技巧：把server_mcp.py和agent_main.py放在同一个 Git 仓库里，用requirements.txt锁定依赖版本。MCP SDK 还在快速迭代，不同版本的 API 可能有细微差异。我遇到过升级 SDK 后FastMCP的导入路径变了，导致 Server 启动失败。锁定版本能避免这类问题。
