> **本教程目标**：用 Python 的 FastMCP 框架，把 `kubectl` 封装成一个让 Claude Code（或任何 MCP 客户端）可以直接调用的 Kubernetes 运维 MCP 服务器。学完你能独立设计"读/写分离 + 安全校验"的工具型 MCP，并用 Inspector 与 Claude Code 联调。 > > **适用人群**：已经会在终端里跑 `kubectl`、有 Python 基础、想把 K8s 运维交给 AI 助手但又怕它乱删集群的 SRE / 后端工程师。 ## 一、为什么是 MCP，为什么是 FastMCP 让大模型操作 Kubernetes，本质是给它一组**可控、可审计、可拒绝**的函数。直接把 `kubectl` 丢给模型去自由发挥是危险的——它可能 `delete ns prod`。MCP（Model Context Protocol）解决的就是"如何把外部能力以**工具 / 资源 / 提示词**三原语的标准方式暴露给模型"，让客户端统一发现、调用、限权。 FastMCP 是 MCP 的 Python 参考框架。它的核心思想是**装饰器即接口**：你写一个普通 Python 函数，加 `@mcp.tool`，它就自动变成符合 MCP 协议、带参数校验和文档的工具。不需要手写 JSON-RPC、不需要手动维护 schema，类型注解直接转成协议字段。 ```mermaid flowchart LR U[你: Claude Code / IDE] -->|自然语言| CC[MCP Client] CC -->|stdio / HTTP| S[FastMCP 服务器] S -->|白名单+校验| K[kubectl] K --> C[(K8s 集群)] S -.读操作只返回文本.-> CC S -.写操作默认 dry-run.-> CC ``` ## 二、前置条件 | 项 | 要求 | 说明 | |---|---|---| | Python | ≥ 3.11 | FastMCP 依赖新特性 | | `uv` | 已安装 | 推荐用 `uv` 跑，免去虚拟环境管理 | | `kubectl` | 已安装且 `kubectl get nodes` 能通 | 服务器本质是包一层 kubectl | | Claude Code | 已安装 | 用作 MCP 客户端联调 | 如果暂时没有真实集群，可以用 `minikube start` 或 `kind create cluster` 起一个本地集群，效果完全一致。 ## 三、第一步：初始化项目 ```bash mkdir k8s-mcp && cd k8s-mcp uv init --python 3.11 uv add fastmcp ``` **预期结果**：目录里出现 `pyproject.toml`、`.python-version`，依赖里多了 `fastmcp`。FastMCP 会自动带上官方 `mcp` SDK，无需重复安装。 **常见问题** - *报错 `fastmcp` 找不到版本*：确认用的是官方包 `fastmcp`（PyPI 上同名），不要装成早期社区 fork。官方仓库在 `gofastmcp.com`。 - *`uv init` 报 Python 版本不可用*：`uv python install 3.11` 后重试。 ## 四、第二步：写第一个只读工具 新建 `kube_mcp.py`： ```python from fastmcp import FastMCP import subprocess mcp = FastMCP("k8s-ops") def _run(args: list[str], timeout: int = 30) -> str: """安全地执行 kubectl 子命令：永不 shell=True，固定超时。""" proc = subprocess.run( ["kubectl", *args], capture_output=True, text=True, timeout=timeout, ) if proc.returncode != 0: return f"ERROR: {proc.stderr.strip()}" return proc.stdout.strip() @mcp.tool def list_pods(namespace: str = "default") -> str: """列出指定命名空间的 Pod，返回 wide 视图。""" return _run(["get", "pods", "-n", namespace, "-o", "wide"]) if __name__ == "__main__": mcp.run(transport="stdio") ``` 几个关键设计点： 1. **`_run` 是唯一的执行出口**，所有工具都走它，方便集中加白名单和限流。 2. **绝不 `shell=True`**：参数以 `list` 传入，从根上杜绝命令注入。 3. **函数 docstring 会变成模型看到的工具说明**——写得清楚，模型才知道何时该用。 先离线验证一下逻辑： ```bash uv run python -c "from kube_mcp import list_pods; print(list_pods())" ``` **预期结果**：打印一张 Pod 表，或一条 `ERROR: ...`（如果 kubeconfig 没配好）。 ## 五、第三步：用 Inspector 联调 MCP 的官方调试利器是 `@modelcontextprotocol/inspector`，它是一个独立 UI，能让你不经过任何大模型、直接手动调用工具，看协议级返回。 ```bash npx -y @modelcontextprotocol/inspector uv run python kube_mcp.py ``` 浏览器打开提示的本地端口 → 左侧 `Tools` → 找到 `list_pods` → 点 `Run Tool`。 **预期结果**：右侧 JSON 返回里能看到 `result.content[0].text` 是 `kubectl get pods` 的输出。 **常见问题** - *Inspector 连不上*：确认传输选 `STDIO`，命令填的是 `uv`，参数是 `run python kube_mcp.py`（分两栏）。 - *中文乱码*：kubectl 输出本身是 UTF-8，乱码多半是终端编码问题，在 MCP 里以 str 传递不影响。 这一步价值很大：**工具在被模型调用前，先用人工验证"接口契约正确"**，避免把 bug 甩锅给模型。 ## 六、第四步：读写分离与安全护栏 运维 MCP 的灵魂是**分级**。读操作（get/describe/logs）随时可跑；写操作（scale/apply/delete）必须默认安全、可回滚。下面把工具补齐： ```python @mcp.tool def describe_resource(kind: str, name: str, namespace: str = "default") -> str: """查看资源详情。kind 例如 pod / deploy / svc。""" return _run(["describe", kind.lower(), name, "-n", namespace]) @mcp.tool def pod_logs(name: str, namespace: str = "default", tail: int = 100) -> str: """获取 Pod 最近 N 行日志。""" return _run(["logs", name, "-n", namespace, f"--tail={tail}"]) @mcp.tool def scale_deployment(name: str, replicas: int, namespace: str = "default") -> str: """扩缩容 Deployment。replicas 会被限制在 0–50。""" if not 0 <= replicas <= 50: return "ERROR: replicas 必须在 0–50 之间" return _run(["scale", "deploy", name, f"--replicas={replicas}", "-n", namespace]) @mcp.tool def apply_manifest(manifest_yaml: str, dry_run: bool = True) -> str: """应用一段 YAML。默认 dry-run=server 仅校验，不真正下发。""" args = ["apply", "-f", "-", *(["--dry-run=server"] if dry_run else [])] proc = subprocess.run( ["kubectl", *args], input=manifest_yaml, capture_output=True, text=True, timeout=30, ) tag = "[DRY-RUN] " if dry_run else "[APPLIED] " return tag + (proc.stdout.strip() or proc.stderr.strip()) ``` 注意三层护栏： - **业务校验**：`scale` 限定副本数范围，挡住模型一时兴起的 `replicas=9999`。 - **默认保守**：`apply` 默认 `dry_run=True`，必须显式传 `dry_run=False` 才真下发。 - **无 `delete` 工具**：危险动作干脆不暴露成工具——MCP 的安全哲学是"不给的能力就用不了"。 > **设计原则**：MCP 工具的命名和参数应该让模型**无法做坏事**，而不是依赖模型自觉。 ## 七、第五步：暴露资源与提示词 工具是"动词"，资源是"名词"，提示词是"剧本"。补上后两个原语： ```python @mcp.resource("k8s://cluster/info") def cluster_info() -> str: """集群控制面地址。""" return _run(["cluster-info"]) @mcp.resource("k8s://nodes") def nodes() -> str: """节点列表，用于排障时定位调度。""" return _run(["get", "nodes", "-o", "wide"]) @mcp.prompt def triage_crash(namespace: str = "default") -> str: """排障剧本：定位并诊断崩溃 Pod。""" return ( f"你是资深 SRE。命名空间 {namespace} 有 Pod 在崩溃。" "请按顺序：1) 调 list_pods 找 CrashLoopBackOff 的 Pod；" "2) 调 describe_resource 看事件；" "3) 调 pod_logs 看 last log；" "4) 给出根因假设和修复建议。每步先说结论再贴证据。" ) ``` `@mcp.resource` 的 URI 是模型读取数据的"地址"，适合那些**变化不频繁、整体可读**的信息；`@mcp.prompt` 把固定的排障 SOP 固化下来，模型调用它就拿到一套连贯指令。 ## 八、第六步：接入 Claude Code 在项目目录跑： ```bash claude mcp add k8s-ops -- uv run python $(pwd)/kube_mcp.py ``` 或直接编辑 `~/.claude.json`（用户级）里的 `mcpServers`： ```json { "mcpServers": { "k8s-ops": { "command": "uv", "args": ["run", "python", "/abs/path/kube_mcp.py"] } } } ``` 验证： ```bash claude mcp list ``` **预期结果**：列表里出现 `k8s-ops ✓ connected`。 然后自然语言驱动： > "k8s-ops，default 命名空间有没有 Pod 不健康？帮我看一下详情和日志。" 模型会自动连调 `list_pods` → `describe_resource` → `pod_logs`，并把结果汇总。`scale` 与 `apply` 默认走 dry-run，模型把"建议执行"的命令贴给你确认。 **常见问题** - *`claude mcp list` 显示 disconnected*：在项目目录 `uv run python kube_mcp.py` 看是否能正常起；常见是 `kubeconfig` 在 Claude Code 的环境里取不到，把 `KUBECONFIG` 写进 server 的 env 配置。 - *模型不调用工具*：工具 docstring 太模糊。重写 docstring，明确"何时用、返回什么"。 ## 九、实战案例：让模型自助排障 CrashLoopBackOff 完整可运行的最小场景： 1. 起一个会崩的 Pod：`kubectl run boom --image=busybox --restart=Always -- /bin/sh -c "exit 1"` 2. 在 Claude Code 里说："用 k8s-ops 的 triage_crash 排查 default 里的崩溃。" 3. 模型按提示词剧本执行，最终输出类似： ``` 根因：容器启动即退出(exit 1)，无业务进程，触发 CrashLoopBackOff。 证据：State=Waiting/Reason=CrashLoopBackOff；上次日志为空。 建议：修正启动命令或换镜像；可先 scale 到 0 止损（需你确认）。 ``` 4. 清理：`kubectl delete pod boom`。 整个排障链路全程可审计——每一步都是一次 MCP 工具调用，记录在 Claude Code 的会话里。 ## 十、延伸阅读 - **FastMCP 官方文档**：`gofastmcp.com`——工具/资源/提示词三原语、传输层、认证的权威说明。 - **MCP 协议规范**：`modelcontextprotocol.io`——理解 JSON-RPC、capabilities、roots 等。 - **官方 Python SDK**：`github.com/modelcontextprotocol/python-sdk`——FastMCP 的底层依赖，可直接看 `mcp.server.fastmcp`。 - **Streamable HTTP 传输**：当你需要把服务器从本机搬到远端给团队共享时，把 `transport="stdio"` 换成 `transport="streamable-http"`，再叠 OAuth 2.1。 ## 十一、训练营适用性评估 - **适合训练营类型**：云原生 / SRE / AIOps 方向的中级实战营；也适合"AI + 运维"主题企业内训。 - **教学时长**：约 3–4 学时。其中框架与三原语 1 学时、读写分离与安全设计 1 学时、Inspector 联调 + Claude Code 集成 1.5 学时。 - **学员产出**：每人一个可联调的 K8s MCP 服务器，并完成一次完整的崩溃 Pod 自助排障演示。 - **进阶作业**：补一个 `rollout history / undo` 工具，并加按命名空间细粒度限权的中间件。 ## 参考来源 1. [FastMCP 官方文档](https://gofastmcp.com/getting-started/welcome) — Python 构建 MCP 服务器的参考框架，三原语与传输层说明。 2. [FastMCP in Python: Build a Real MCP Server (2026 Guide)](https://www.danilchenko.dev/posts/fastmcp-mcp-server/) — 生产级 MCP 服务器构建教程，含工具/资源/提示词与 OAuth 示例。 3. [MCP Server Tutorial: 12 Steps Python FastMCP (2026)](https://tech-insider.org/mcp-server-tutorial-python-fastmcp-claude-2026/) — 12 步从零搭建并与 Claude 集成的实操指南。 4. [k8s-mcp-server (GitHub)](https://github.com/adityajoshi12/k8s-mcp-server) — 社区 K8s MCP 服务器参考实现，FastMCP + kubectl 思路对照。 5. [Model Context Protocol 官方站点](https://modelcontextprotocol.io/) — MCP 协议规范、Inspector 工具与各语言 SDK 入口。 据库超时"，然后用自然语言告诉你根因。整个过程你只敲了一句话——这就是 MCP 把"重复运维动作"变成"自然语言接口"的威力。 进阶玩法：再给它加一个 `restart_container` 工具（带 `deny` 默认禁用、需手动放行），就能让模型在确认后帮你重启卡死的容器，形成"诊断 → 自愈"闭环。 ## 工作流程全景 ```mermaid sequenceDiagram participant U as 用户 participant CC as Claude Code participant DP as docker-patrol MCP participant D as docker CLI U->>CC: 哪些容器不健康？ CC->>DP: tools/call patrol_all DP->>D: docker ps -a --format json D-->>DP: 容器状态列表 DP-->>CC: 异常分级 JSON CC->>DP: tools/call inspect_container(api-server) DP->>D: docker inspect + logs D-->>DP: 元数据+日志 DP-->>CC: 聚合诊断 CC-->>U: "api-server unhealthy，日志显示DB超时" ``` ## 延伸阅读 - **官方 MCP 文档** [modelcontextprotocol.io](https://modelcontextprotocol.io/docs/getting-started/intro) ——协议规范、所有原语与传输类型的权威说明 - **Claude Code MCP 接入文档** ——transport、tool search、鉴权、错误处理的官方指引 - **FastMCP 进阶** ——异步工具、上下文对象、资源订阅等更高级用法 - 把本服务器装进 Docker 镜像做成"远程 HTTP MCP"，参考文末"Run MCP Server in Docker" ## 训练营适用性评估 | 维度 | 评估 | |------|------| | 适合训练营类型 | DevOps/SRE 方向、Python 后端、AI 工程化训练营 | | 教学时长 | 约 2.5 小时（0.5h 原理 + 1.5h 编码注册 + 0.5h 调试护栏） | | 学员基础要求 | Python 基础 + 用过 docker，不要求懂 MCP 协议细节 | | 可交付物 | 一个可复用的只读巡检 MCP + 团队级 `.mcp.json` 模板 | | 拓展作业 | 加 `restart_container` 写工具并配 deny 护栏；改用 HTTP 传输部署成远程 MCP | ## 参考来源 1. [Connect to external tools with MCP — Claude Code Docs](https://code.claude.com/docs/en/agent-sdk/mcp) — 官方 MCP 接入与配置（transport、scope、鉴权、错误处理）权威文档 2. [How to Build an MCP Server with Python, Docker, and Claude Code — freeCodeCamp](https://www.freecodecamp.org/news/how-to-build-an-mcp-server-with-python-docker-and-claude-code/) — FastMCP + Docker + Claude Code 全流程实战范例 3. [Claude Code MCP Server Setup: Enterprise Integration Guide — Dextra Labs](https://dextralabs.com/blog/claude-code-mcp-enterprise-ai-integrations/) — 企业级 MCP 集成、`claude mcp add` 与多服务器管理 4. [Automate Docker Workflows with Claude Code + MCP (2026) — MCSA Guru](https://mcsaguru.com/automate-docker-workflows-claude-code-mcp) — 用 MCP 巡检容器、读日志、排查 Docker 的同主题参考 5. [Run MCP Server in Docker — Itbusina](https://itbusina.com/blog/run-mcp-server-docker/) — 把 MCP 服务器容器化、改用远程传输的延伸做法 > [!tip] 51学通信知识星球 > **星主：爱卫生**（微信：gprshome201101）——公众号「51学通信」创办者，19 年通信行业运维与讲师经验，著有《GPRS网络信令实例详解》《EPC原理与实践》《5G核心网原理与实践》三本纸质书（各大电商平台有售）。 > **星球定位**：深耕 5G 核心网、IMS、NFV/AI、云原生等 ICT 领域，主打运维岗位与信令分析。 > **当前规模**：5200 篇主题 / 990 个回答，198 元/年。 > **加入福利**：加入星球后请加爱卫生微信，拉您进知识星球年度陪伴与答疑群——不限量回复您的各种问题，另有群内分享。 > **AI 加持**：51学通信知识星球已作为 skill 接入 WorkBuddy，使用 deepseek-4-flash 在海量星球帖子中精准找贴、找视频、找 FAQ 和案例，做总结、找文档和 PPT 等场景全覆盖。