摘要：在 Linux/Ubuntu 运维与本地开发中，服务启动失败（Failed/Crash）是高频痛点。传统的排查流程往往是在终端反复敲击 systemctl status、journalctl -xeu、dmesg 和 ss -tulpn。本文基于 Anthropic 的 MCP（Model Context Protocol） 协议与 Python FastMCP，带你从零构建一个专属于 Linux 的 Systemd 与内核日志智能诊断 MCP Server，让 AI Agent 自主拉取日志、定位端口冲突与依赖缺失，并完成一键自愈。

一、 为什么要做 Systemd 诊断 MCP？

在日常开发或服务器运维中，后台守护进程崩溃的原因五花八门：端口被占、环境变量缺失、配置语法错误、段错误（Segmentation fault）或内核 OOM。

  • 传统排障痛点：

     

    • 命令冗长碎片化（journalctl -u app --no-pager -n 50）。

    • 报错信息分散在 Systemd 日志与内核 ring buffer（dmesg）中，容易遗漏内核级崩溃。

    • 遇到“端口占用”或“缺动态链接库”，新手仍需多步定位占用进程 PID 或排查 ldd。

  • MCP 赋能后的体验：

    你只需在终端或 IDE 中输入：

     

    “我的 nginx 和后端 my-app 状态异常，帮我查下挂掉的原因，如果只是端口冲突把占用的进程关掉并重启服务。”

Agent 会自动按需调用底层封装好的工具链，完成“抓取日志 $\rightarrow$ 分析根因 $\rightarrow$ 执行修复 $\rightarrow$ 验证状态”的全流程。

二、 架构与功能设计

该 MCP Server 运行在本地宿主机（或通过 SSH 运行在远程机），暴露以下四大核心能力：

  1. 服务生命周期控制 (manage_service)：封装 systemctl start/stop/restart/reload/status。

  2. 精准日志抓取 (get_journal_logs)：支持按服务名、日志级别（err/warning）、行数快速提取 journalctl。

  3. 内核级事件排查 (inspect_kernel_dmesg)：抓取 dmesg 中的 OOM Killer、段错误、硬件/驱动异常。

  4. 端口与网络冲突检测 (check_port_conflict)：封装 ss/lsof，快速找出指定端口当前的监听者。

三、 实战：构建 Systemd 诊断 MCP Server

1. 环境准备

推荐使用 Python 3.10+，安装 FastMCP SDK 与 Pydantic：

pip install mcp pydantic

确保当前运行环境对目标服务具备免密 sudo 权限（或在测试用户态服务 systemctl --user 时直接使用）。

2. 完整实现代码 (systemd_mcp_server.py)

#!/usr/bin/env python3
"""
Systemd & Linux Log Diagnostics MCP Server
提供 Systemd 服务管理、Journal 日志分析、内核 dmesg 检查及端口冲突诊断能力。
"""

import asyncio
import subprocess
from typing import Optional, List
from mcp.server.fastmcp import FastMCP

mcp = FastMCP("systemd-diagnostics-server")


async def run_shell(cmd: List[str], timeout: int = 15) -> str:
    """异步安全执行本地命令"""
    try:
        proc = await asyncio.create_subprocess_exec(
            *cmd,
            stdout=asyncio.subprocess.PIPE,
            stderr=asyncio.subprocess.PIPE,
        )
        stdout, stderr = await asyncio.wait_for(proc.communicate(), timeout=timeout)
        
        out_str = stdout.decode("utf-8", errors="replace").strip()
        err_str = stderr.decode("utf-8", errors="replace").strip()
        
        result = []
        if out_str:
            result.append(out_str)
        if err_str:
            result.append(f"[STDERR]\n{err_str}")
        if proc.returncode != 0:
            result.append(f"[EXIT CODE] {proc.returncode}")
            
        return "\n".join(result) if result else "执行成功（无输出内容）"
    except asyncio.TimeoutError:
        return f"Error: 命令执行超时（限时 {timeout}s）"
    except Exception as e:
        return f"执行失败: {str(e)}"


# ==================== MCP 工具定义 ====================

@mcp.tool()
async def manage_service(service_name: str, action: str, is_user_mode: bool = False) -> str:
    """
    管理 Systemd 服务的状态。
    :param service_name: 服务名称（例如: nginx, docker, my-app）
    :param action: 动作，支持: status, start, stop, restart, reload, is-active
    :param is_user_mode: 是否为用户态服务（--user）
    """
    valid_actions = ["status", "start", "stop", "restart", "reload", "is-active"]
    if action not in valid_actions:
        return f"Error: 不支持的操作，仅支持: {', '.join(valid_actions)}"

    cmd = ["systemctl"]
    if is_user_mode:
        cmd.append("--user")
    
    cmd.extend([action, service_name, "--no-pager"])
    return await run_shell(cmd)


@mcp.tool()
async def get_journal_logs(
    service_name: Optional[str] = None,
    lines: int = 50,
    priority: Optional[str] = None,
    since: Optional[str] = None,
    is_user_mode: bool = False,
) -> str:
    """
    从 journalctl 中提取详细的运行日志与堆栈报错。
    :param service_name: 目标服务单元名（可选，如传空则拉取系统级关键日志）
    :param lines: 拉取的日志行数（默认 50）
    :param priority: 日志级别过滤（如: 'err', 'warning', 'info'）
    :param since: 时间范围（如: '10 min ago', 'today', '2026-03-01'）
    :param is_user_mode: 是否为用户态服务日志
    """
    cmd = ["journalctl", "--no-pager", "-n", str(lines)]
    
    if is_user_mode:
        cmd.append("--user")
    if service_name:
        cmd.extend(["-u", service_name])
    if priority:
        cmd.extend(["-p", priority])
    if since:
        cmd.extend(["--since", since])

    return await run_shell(cmd)


@mcp.tool()
async def inspect_kernel_dmesg(lines: int = 40, grep_pattern: Optional[str] = None) -> str:
    """
    读取 Linux 内核环形缓冲区日志（dmesg），用于诊断 OOM Killer 杀进程、硬件故障或驱动崩溃。
    :param lines: 输出行数
    :param grep_pattern: 关键词过滤（如 'oom', 'segfault', 'kill'）
    """
    cmd = ["dmesg", "-T", "--level=emerg,alert,crit,err,warn"]
    output = await run_shell(cmd)
    
    lines_list = output.splitlines()
    if grep_pattern:
        lines_list = [line for line in lines_list if grep_pattern.lower() in line.lower()]
    
    selected_lines = lines_list[-lines:] if len(lines_list) > lines else lines_list
    return "\n".join(selected_lines) if selected_lines else "未检测到相关内核报警日志。"


@mcp.tool()
async def check_port_conflict(port: int) -> str:
    """
    检查指定端口是否被占用，并返回占用进程的 PID、进程名及监听地址。
    :param port: 要查询的端口号（如 80, 8080, 3306）
    """
    cmd = ["ss", "-tulpn"]
    output = await run_shell(cmd)
    
    matched = [line for line in output.splitlines() if f":{port} " in line or f":{port}\t" in line]
    if not matched:
        return f"端口 {port} 当前处于空闲状态，未检测到占用冲突。"
    
    header = "Netid  State   Recv-Q  Send-Q   Local Address:Port   Peer Address:Port  Process"
    return f"{header}\n" + "\n".join(matched)


if __name__ == "__main__":
    # 以标准 I/O (stdio) 方式运行
    mcp.run(transport="stdio")

四、 接入与实战场景演练

1. 在 MCP Client 中配置

在 Claude Desktop 或其他支持 MCP 规范的 Agent 配置文件中注册该 Server：

{
  "mcpServers": {
    "systemd-diag": {
      "command": "python3",
      "args": [
        "/home/ubuntu/scripts/systemd_mcp_server.py"
      ]
    }
  }
}

2. 真实排障对话示例

  • 场景：服务启动报错 203/EXEC 或端口占用

     

    用户：“我的 api-gateway 服务刚重启失败了，帮忙查一下原因。”

     

    Agent 自动执行流程：

     

    1. 调用 manage_service(service_name="api-gateway", action="status")，识别出 Active: failed (Result: exit-code)。

    2. 调用 get_journal_logs(service_name="api-gateway", lines=30, priority="err")，抓取到底层报错 bind: address already in use :8080。

    3. 调用 check_port_conflict(port=8080)，发现旧的 Java 孤儿进程（PID 14208）未释放。

    4. 输出排障结论：“由于遗留的 PID 14208 占用了 8080 端口，导致新服务无法绑定地址。建议杀死该进程后重新执行 systemctl restart api-gateway。”

  • 场景：进程被系统静默杀死（OOM 排查）

     

    用户：“后台的深度学习推理服务突然消失了，没有生成 core dump，发生了什么？”

     

    Agent 自动执行流程：

     

    1. 调用 inspect_kernel_dmesg(grep_pattern="oom")。

    2. 从日志中精准捕获 Out of memory: Killed process 38921 (python3)。

    3. 输出排障结论：“检测到系统在 10:14:22 触发了 Linux OOM Killer，已强制终止了物理内存占比最高的 python3 进程。建议调整 batch size 或配置 swap 交换区。”

五、 生产环境优化建议

  1. Sudo 权限最小化配置：

    在 /etc/sudoers.d/agent-ops 中限制免密命令范围，避免直接使用无限制的 root 执行环境：

    ubuntu ALL=(ALL) NOPASSWD: /usr/bin/systemctl status *, /usr/bin/systemctl restart *, /usr/bin/journalctl *

  2. Prompts 模版联动：

    可以在 FastMCP 中注册专用的 @mcp.prompt() 模版（如 systemd_triage），将标准化的 Linux 排障 SOP 固化为 Agent 的系统提示词，大幅提升推理准确率。

# linux # ubuntu # 运维
Logo
AI Agent技术社区

Agent 垂直技术社区，欢迎活跃、内容共建。

更多推荐

cover

程序员学 AI（八）：AI Agent 到底是什么？从回答问题到完成任务

avatar AI Agent技术社区
cover

22道AI Agent 工程师必会的面试题，别啥都不会就出去找工作

avatar AI Agent技术社区

写给 TypeScript 工程师的 OpenClaw.NET 上手指南：用你熟悉的 TS 思维，跑起一个生产级 AI Agent

读代码层：TS 和 C# 本来就是「远房亲戚」——asyncawait??、接口、泛型、严格空检查，你的心智模型几乎无缝平移。真正要适应的只有两件事：类型是具名而非结构化的（多一句显式声明），以及 NativeAOT 下的「禁反射」约束。写扩展层：你甚至可以不怎么碰 C#——Node.js 插件桥意味着你的 TypeScript 插件就是一等公民，而且能在生产 aot 车道跑。内核给你 .NET

avatar AI Agent技术社区