摘要：传统的服务器运维面板（如宝塔面板）虽然直观，但操作仍停留在“点点点”的阶段，遇到复杂故障排查时灵活性明显不足。本文将带你基于 Anthropic 推出的 MCP（Model Context Protocol） 规范与 Python FastMCP SDK，从零构建一个可通过 SSH 管理多台远程服务器的 MCP Server，实现容器管理、Nginx 站点配置、系统监控与一键排障的“对话式运维面板”。

一、 为什么要做基于 MCP 的运维面板？

传统面板（宝塔）vs MCP 智能运维

  • 传统 Web 面板：功能全面、可视化强，但需要在浏览器多层菜单中跳转，复杂任务（如定位高负载来源并结合日志自动修复）需要人工介入。

  • MCP 运维 Agent：将服务器核心管理能力抽象为标准 Tool 与 Resource，通过大语言模型直接调用。你只需在 IDE 或终端敲下一句话：

     

    “帮我检查 192.168.1.50 的 Nginx 状态，如果 80 端口挂了尝试重启，并把错误日志最后 20 行输出给我。”

Agent 即可自主规划步骤、调用 SSH 驱动执行并反馈结果。

二、 架构设计与交互流程

整体架构由 MCP Host（Claude Desktop / CLI 工具 / 终端 Agent） 与 自定义 SSH MCP Server 组成：

  1. Host 层：负责接收自然语言指令，结合上下文规划 Tool 调用链。

  2. Server 层（我们的代码）：提供基于 asyncssh 的异步连接池，并暴露格式化的运维工具集。

  3. Target 远程机：无需部署臃肿的 Web 守护进程，仅通过标准 SSH 协议（基于密钥认证）完成安全交互。

Plaintext

[ 开发者自然语言 ] 
       │
       ▼
[ MCP Host (Claude/Agent) ] ──(stdio/JSON-RPC)──> [ SSH MCP Server ]
                                                        │ (SSH/SFTP)
                                                        ▼
                                              [ 远程 Linux 服务器 ]

三、 实战：构建全功能运维 MCP Server

1. 环境准备与依赖安装

建议使用 Python 3.10+ 环境，安装官方 FastMCP SDK 与异步 SSH 库：


Bash

pip install mcp asyncssh pydantic

2. 核心源码：ops_mcp_server.py

创建一个包含命令执行、监控指标、Docker容器管理与Nginx运维的完整服务：


Python

#!/usr/bin/env python3
"""
Remote Server Ops MCP Server
基于 FastMCP 实现的自然语言远程服务器运维管理服务
"""

import os
import asyncio
from typing import Optional, Dict
import asyncssh
from mcp.server.fastmcp import FastMCP

# 初始化 MCP Server
mcp = FastMCP("remote-ops-manager")

# 全局连接池缓存：避免每次工具调用都重新握手 SSH
_SSH_POOL: Dict[str, asyncssh.SSHClientConnection] = {}


async def get_connection(
    host: str,
    username: str,
    port: int = 22,
    private_key_path: Optional[str] = None,
) -> asyncssh.SSHClientConnection:
    """获取或复用 SSH 连接池中的连接"""
    key = f"{username}@{host}:{port}"
    conn = _SSH_POOL.get(key)
    
    # 检查连接是否可用
    if conn and not conn.is_closed():
        return conn

    client_keys = []
    if private_key_path:
        client_keys.append(os.path.expanduser(private_key_path))
    elif os.path.exists(os.path.expanduser("~/.ssh/id_rsa")):
        client_keys.append(os.path.expanduser("~/.ssh/id_rsa"))

    new_conn = await asyncssh.connect(
        host=host,
        port=port,
        username=username,
        client_keys=client_keys,
        known_hosts=None, # 实际生产环境可配置 known_hosts
    )
    _SSH_POOL[key] = new_conn
    return new_conn


async def run_remote_sh(
    host: str, username: str, command: str, port: int = 22, timeout: int = 30
) -> str:
    """底层安全执行 Shell 脚本"""
    # 基础高危指令拦截沙箱
    dangerous_keywords = ["rm -rf /", "mkfs", "dd if=/dev/zero", ":(){:|:&};:"]
    if any(kw in command for kw in dangerous_keywords):
        return "Error: 检测到高危系统指令，执行被安全策略阻断！"

    try:
        conn = await get_connection(host=host, username=username, port=port)
        result = await asyncio.wait_for(conn.run(command), timeout=timeout)
        
        output = []
        if result.stdout:
            output.append(f"[STDOUT]\n{result.stdout.strip()}")
        if result.stderr:
            output.append(f"[STDERR]\n{result.stderr.strip()}")
        output.append(f"[STATUS] Exit code: {result.exit_status}")
        return "\n".join(output)
    except asyncio.TimeoutError:
        return f"Error: 执行超时（限时 {timeout}s）"
    except Exception as e:
        return f"SSH 通信异常: {str(e)}"


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

@mcp.tool()
async def execute_raw_command(
    host: str, username: str, command: str, port: int = 22
) -> str:
    """在远程服务器上直接执行任意 Linux Shell 命令并返回结果"""
    return await run_remote_sh(host, username, command, port)


@mcp.tool()
async def get_server_dashboard(
    host: str, username: str, port: int = 22
) -> str:
    """获取服务器类似宝塔首页的综合健康度看板（CPU、内存、磁盘、负载、Top5进程）"""
    cmd = (
        "echo '=== 1. 负载与运行时间 ==='; uptime; "
        "echo -e '\n=== 2. 内存使用概览 (MB) ==='; free -m; "
        "echo -e '\n=== 3. 磁盘挂载状态 ==='; df -h -x tmpfs -x devtmpfs; "
        "echo -e '\n=== 4. 高占用进程 TOP 5 ==='; ps -eo pid,user,%cpu,%mem,cmd --sort=-%cpu | head -n 6"
    )
    return await run_remote_sh(host, username, cmd, port)


@mcp.tool()
async def manage_docker_containers(
    host: str, username: str, action: str, container_name: Optional[str] = None, port: int = 22
) -> str:
    """
    管理远程机器的 Docker 容器。
    :param action: 支持 'list' (查看列表), 'start', 'stop', 'restart', 'logs' (查看日志)
    :param container_name: 目标容器名称或ID（当 action 为 start/stop/restart/logs 时必填）
    """
    if action == "list":
        cmd = "docker ps -a --format 'table {{.ID}}\t{{.Names}}\t{{.Status}}\t{{.Ports}}'"
    elif action in ["start", "stop", "restart"]:
        if not container_name:
            return "Error: 操作容器必须指定 container_name"
        cmd = f"docker {action} {container_name}"
    elif action == "logs":
        if not container_name:
            return "Error: 查看日志必须指定 container_name"
        cmd = f"docker logs --tail 50 {container_name}"
    else:
        return f"Error: 不支持的 Docker 动作: {action}"

    return await run_remote_sh(host, username, cmd, port)


@mcp.tool()
async def manage_nginx(
    host: str, username: str, action: str, port: int = 22
) -> str:
    """
    管理远程 Nginx 服务。
    :param action: 支持 'status' (查看状态), 'test' (检查配置语法), 'reload' (平滑重载), 'restart' (重启)
    """
    actions_map = {
        "status": "systemctl status nginx --no-pager",
        "test": "nginx -t",
        "reload": "nginx -t && systemctl reload nginx",
        "restart": "systemctl restart nginx",
    }
    if action not in actions_map:
        return f"Error: 仅支持的操作: {list(actions_map.keys())}"
    
    return await run_remote_sh(host, username, actions_map[action], port)


if __name__ == "__main__":
    # 以标准 I/O 运行服务，便于 Host 进程管理
    mcp.run(transport="stdio")

四、 接入与测试

1. 配置 MCP Host（以 Claude Desktop 为例）

编辑你的客户端配置文件（macOS 为 ~/Library/Application Support/Claude/claude_desktop_config.json，Linux/Windows 为对应配置路径）：


JSON

{
  "mcpServers": {
    "remote-ops": {
      "command": "python3",
      "args": [
        "/home/your_user/ops_mcp_server.py"
      ]
    }
  }
}

2. 实际交互效果示例

  • 场景一：综合巡检

     

    你：“查看一下 10.0.0.8 上的服务器状态，看有没有资源告警。”

    Agent：自动调用 get_server_dashboard，解析内存余量与 CPU Top 进程，并用表格输出结论。

  • 场景二：容器运维排障

     

    你：“10.0.0.8 上跑的 redis 容器好像断了，帮我重启一下并看下最新的日志。”

    Agent：依次调用 manage_docker_containers(action='restart', container_name='redis') 与 manage_docker_containers(action='logs', container_name='redis')，输出执行结果与关键日志。

五、 生产级扩展考量

  1. 安全审计与最小权限（Sudoers）：

    建议在远程目标机上配置专用的低权限运维账户（如 agent_ops），并通过 /etc/sudoers 精确配置免密执行的白名单命令（如 systemctl reload nginx、docker 等），避免使用 root 直连。

  2. 多节点拓扑管理：

    可引入本地 YAML/SQLite 配置，将 IP 映射为语义标签（如 prod-web-01、staging-db），让大模型直接基于标签分发指令。

# 前端 # 运维
Logo
DeepSeek技术社区

欢迎加入DeepSeek 技术社区。在这里，你可以找到志同道合的朋友，共同探索AI技术的奥秘。

更多推荐

cover

人均557万，寒武纪6.9亿员工「大红包」将到账；格力电器开办技校，董明珠亲自当校长；总榜第2，DeepSeek V4 Pro中文测评出炉 | 极客头条

avatar DeepSeek技术社区
cover

魔珐星云 + DeepSeek：我给女儿搭了个会辅导作业的 3D 数字人老师

avatar DeepSeek技术社区
cover

DeepSeek + 魔珐星云：10 分钟打造一个会望闻问切的 AI 健康顾问数字人

avatar DeepSeek技术社区