基于 Python + MCP + LLM 手搓智能服务器运维助手
为我们提供了一种标准化的方式来连接 LLM 和本地基础设施，而 Python 则是连接这一切的胶水。这种“手搓”的方案虽然前期代码量稍大，但它赋予了开发者最大的自由度——你可以在任何地方运行它，无论是树莓派、本地笔记本，还是云服务器，这才是真正的“应用落地”。未来，我们可以引入 LangGraph 来管理更复杂的状态，或者让多个 MCP Agent 互相协作，真正实现无人值守的 AIOps。
摘要：传统运维监控（如 Zabbix、Prometheus）最大的痛点在于“告警风暴”与“语义缺失”。当 CPU 飙升或日志报错时，运维人员往往只收到冷冰冰的指标，却不知道“发生了什么”以及“怎么修”。本文将探讨如何利用 MCP (Model Context Protocol) 开放协议，结合 Python 与 LLM（如 DeepSeek/OpenAI），从零构建一个能够通过自然语言分析日志、查阅本地知识库并给出修复建议的 Agent。
关键词：MCP, 智能体, 运维自动化, Python, LangChain, 架构设计
1. 为什么我们需要一个 AI 运维智能体？
作为一名开发者，最怕的就是凌晨三点手机震动——服务器 CPU 报警了。通常我们爬起来打开电脑，流程是固定的：
-
SSH 连上服务器。
-
输入
top查看进程。 -
输入
tail -f error.log查看报错日志。 -
去内网 Wiki 搜索这个错误代码以前有没有出现过。
-
尝试重启或回滚。
这个过程高度标准化，却极度消耗精力。现在的 LLM（大模型）已经具备了极强的代码理解和推理能力，那么，为什么不能让 AI 替我完成前 4 步，直接把“诊断报告”发给我呢？
不依赖任何封闭的低代码平台，我们完全可以使用 MCP (Model Context Protocol) 标准和 Python 编写一个独立运行的 Agent，让它成为我们的运维助手。
2.去平台化的纯代码方案
为了保证系统的可移植性和掌控力，我们采用“纯代码”模式构建。
-
工具层 (MCP Server)：基于 Model Context Protocol 标准，将服务器操作（读日志、查状态）封装为标准工具接口。
-
大脑层 (Agent Core)：使用 Python (配合 LangChain 或 原生 API) 编写控制逻辑，负责思考和调度。
-
知识层 (RAG)：使用轻量级向量库（如 ChromaDB）或简单的本地文档加载器，存储运维手册。
逻辑架构
graph TD
A[监控系统/定时任务] -->|Webhook/CLI 触发| B(Python Agent 主程序)
subgraph 智能体核心逻辑
B --> C{需要获取信息?}
C -->|是| D[MCP Client]
D -->|Stdio/SSE 协议调用| E[本地 MCP Server]
E -->|执行 top/tail 命令| E1[Linux Shell]
E1 -->|返回数据| D
D --> F[本地知识库检索]
F -->|注入排查手册| G[LLM (DeepSeek/OpenAI)]
end
G -->|输出诊断报告| H[钉钉/飞书 API]
3. 构建 MCP 运维工具服务 (Server 端)
MCP 是 Anthropic 开源的一个标准协议，它让 AI 连接外部工具变得极其简单且标准。我们首先编写一个运维工具箱。
我们需要编写一个轻量级的 MCP Server，提供两个核心工具：
-
get_server_status: 获取当前 CPU 和 内存状态。 -
read_error_logs: 读取指定行数的错误日志。
以下是基于 Python SDK 的完整实现代码：
# server_mcp.py
# 这是一个标准的 MCP Server，任何支持 MCP 的客户端都能调用它
import sys
import psutil
import subprocess
from mcp.server.fastmcp import FastMCP
# 初始化 MCP 服务，命名为 OpsWatchdog
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:
# 使用 tail 命令读取日志，避免读取大文件卡死
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__":
# 启动 MCP 服务
print("MCP Server Running...", file=sys.stderr)
mcp.run()
4. 构建智能体控制流 (Client 端)
这是“脱离平台”的关键。我们需要自己编写一段 Python 代码来作为“客户端”，连接上面的 MCP Server，并把工具暴露给 LLM。
这里我们模拟一个简单的 Agent 思考循环（ReAct Loop）：
# agent_main.py
import asyncio
from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client
from openai import OpenAI # 可以替换为任何兼容 OpenAI 格式的客户端，如 DeepSeek
# 配置 LLM
client = OpenAI(api_key="your-api-key", base_url="[https://api.deepseek.com](https://api.deepseek.com)")
async def run_agent():
# 1. 启动并连接到我们的 MCP Server
server_params = StdioServerParameters(
command="python", # 使用 python 运行
args=["server_mcp.py"], # 刚才写的 server 文件
)
async with stdio_client(server_params) as (read, write):
async with ClientSession(read, write) as session:
# 2. 初始化：获取 MCP Server 提供的所有工具
await session.initialize()
tools = await session.list_tools()
# 3. 构建 Prompt，告诉 LLM 有哪些工具可用
tool_descriptions = "\n".join([f"- {t.name}: {t.description}" for t in tools.tools])
system_prompt = f"""
你是一个资深的 SRE 运维专家。你可以使用以下工具来排查服务器问题：
{tool_descriptions}
当收到报警时，请先调用工具获取信息，然后分析原因。
"""
# 4. 模拟收到报警场景
user_query = "服务器 CPU 突然飙升，请排查原因并给出建议。"
print(f"用户指令: {user_query}")
# --- 这里简化了 ReAct 循环，直接演示一次工具调用 ---
# 第一轮：询问 LLM 想干什么
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}")
# 假设 LLM 决定调用 get_server_status (在真实代码中需要解析 JSON)
if "get_server_status" in llm_thought:
print(">> Agent 决定调用工具: get_server_status")
tool_result = await session.call_tool("get_server_status")
print(f">> 工具返回结果: {tool_result.content[0].text}")
# 第二轮：将工具结果喂回给 LLM 进行最终诊断
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"工具执行结果: {tool_result.content[0].text}"}
]
)
print(f"Agent 最终诊断: {final_response.choices[0].message.content}")
if __name__ == "__main__":
asyncio.run(run_agent())
5. 本地知识库与 RAG 融合
只有数据（日志）没有知识（经验），AI 只能告诉你“报错了”，不能告诉你“怎么修”。
为了增强 Agent，我们在本地实现一个简单的 RAG（检索增强生成）。
我们可以准备一个 knowledge.txt，里面存放历史故障记录：
[知识库片段]
- 如果发现 CPU 高且进程为 image_process，通常是图片死循环，建议 kill 进程。
- Error 502 通常是 PHP-FPM 进程数不足，建议修改 php.ini max_children。
在 agent_main.py 中，我们可以在发送给 LLM 之前，先简单读取这个文件，将其内容作为 Context 拼接到 System Prompt 中。
对于更复杂的场景，可以使用 chromadb 库将文档向量化，实现语义搜索。
6. 效果演示与价值分析
通过上述纯代码实现，我们拥有了一个完全可控的智能体。
场景模拟：
假设服务器因为某个死循环代码导致 CPU 飙升。
-
运行 Agent：
python agent_main.py
-
控制台输出：
用户指令: 服务器 CPU 突然飙升，请排查原因并给出建议。 Agent 思考: 我需要先查看当前的服务器状态，确认 CPU 使用率。调用 get_server_status。 >> Agent 决定调用工具: get_server_status >> 工具返回结果: CPU使用率: 98.5% Agent 最终诊断: **[高危警告]** 检测到服务器 CPU 使用率高达 98.5%。 结合知识库判断，可能是计算密集型任务卡死。建议立即 SSH 登录使用 top 命令定位具体 PID，并考虑执行 kill 操作。
总结：
-
零成本部署：不需要购买昂贵的 AI 平台订阅，本地 Python 环境即可运行。
-
数据隐私：除了调用 LLM API，所有的日志数据都在本地流转，通过 MCP 安全隔离。
-
无限扩展：你可以随意编写新的 Python 函数（如
restart_service），只需加一行@mcp.tool()，Agent 就能立刻学会新技能。
7. 总结与展望
Model Context Protocol (MCP) 为我们提供了一种标准化的方式来连接 LLM 和本地基础设施，而 Python 则是连接这一切的胶水。这种“手搓”的方案虽然前期代码量稍大，但它赋予了开发者最大的自由度——你可以在任何地方运行它，无论是树莓派、本地笔记本，还是云服务器，这才是真正的“应用落地”。
未来，我们可以引入 LangGraph 来管理更复杂的状态，或者让多个 MCP Agent 互相协作，真正实现无人值守的 AIOps。
有“AI”的1024 = 2048，欢迎大家加入2048 AI社区
更多推荐
14
0- 0
已为社区贡献5条内容
所有评论(0)