本教程目标：用 Python 的 FastMCP 框架，把
kubectl封装成一个让 Claude Code（或任何 MCP 客户端）可以直接调用的 Kubernetes 运维 MCP 服务器。学完你能独立设计"读/写分离 + 安全校验"的工具型 MCP，并用 Inspector 与 Claude Code 联调。适用人群：已经会在终端里跑
kubectl、有 Python 基础、想把 K8s 运维交给 AI 助手但又怕它乱删集群的 SRE / 后端工程师。
让大模型操作 Kubernetes，本质是给它一组可控、可审计、可拒绝的函数。直接把 kubectl 丢给模型去自由发挥是危险的——它可能 delete ns prod。MCP（Model Context Protocol）解决的就是"如何把外部能力以工具 / 资源 / 提示词三原语的标准方式暴露给模型"，让客户端统一发现、调用、限权。
FastMCP 是 MCP 的 Python 参考框架。它的核心思想是装饰器即接口：你写一个普通 Python 函数，加 @mcp.tool，它就自动变成符合 MCP 协议、带参数校验和文档的工具。不需要手写 JSON-RPC、不需要手动维护 schema，类型注解直接转成协议字段。
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 起一个本地集群，效果完全一致。
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：
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")
几个关键设计点：
_run 是唯一的执行出口，所有工具都走它，方便集中加白名单和限流。shell=True：参数以 list 传入，从根上杜绝命令注入。先离线验证一下逻辑：
uv run python -c "from kube_mcp import list_pods; print(list_pods())"
预期结果：打印一张 Pod 表，或一条 ERROR: ...（如果 kubeconfig 没配好）。
MCP 的官方调试利器是 @modelcontextprotocol/inspector，它是一个独立 UI，能让你不经过任何大模型、直接手动调用工具，看协议级返回。
npx -y @modelcontextprotocol/inspector uv run python kube_mcp.py
浏览器打开提示的本地端口 → 左侧 Tools → 找到 list_pods → 点 Run Tool。
预期结果：右侧 JSON 返回里能看到 result.content[0].text 是 kubectl get pods 的输出。
常见问题
STDIO，命令填的是 uv，参数是 run python kube_mcp.py（分两栏）。这一步价值很大：工具在被模型调用前，先用人工验证"接口契约正确"，避免把 bug 甩锅给模型。
运维 MCP 的灵魂是分级。读操作（get/describe/logs）随时可跑；写操作（scale/apply/delete）必须默认安全、可回滚。下面把工具补齐：
@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 工具的命名和参数应该让模型无法做坏事，而不是依赖模型自觉。
工具是"动词"，资源是"名词"，提示词是"剧本"。补上后两个原语：
@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 mcp add k8s-ops -- uv run python $(pwd)/kube_mcp.py
或直接编辑 ~/.claude.json（用户级）里的 mcpServers：
{
"mcpServers": {
"k8s-ops": { "command": "uv", "args": ["run", "python", "/abs/path/kube_mcp.py"] }
}
}
验证：
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 配置。完整可运行的最小场景：
kubectl run boom --image=busybox --restart=Always -- /bin/sh -c "exit 1"根因：容器启动即退出(exit 1)，无业务进程，触发 CrashLoopBackOff。
证据：State=Waiting/Reason=CrashLoopBackOff；上次日志为空。
建议：修正启动命令或换镜像；可先 scale 到 0 止损（需你确认）。
kubectl delete pod boom。整个排障链路全程可审计——每一步都是一次 MCP 工具调用，记录在 Claude Code 的会话里。
gofastmcp.com——工具/资源/提示词三原语、传输层、认证的权威说明。modelcontextprotocol.io——理解 JSON-RPC、capabilities、roots 等。github.com/modelcontextprotocol/python-sdk——FastMCP 的底层依赖，可直接看 mcp.server.fastmcp。transport="stdio" 换成 transport="streamable-http"，再叠 OAuth 2.1。rollout history / undo 工具，并加按命名空间细粒度限权的中间件。进阶玩法：再给它加一个 restart_container 工具（带 deny 默认禁用、需手动放行），就能让模型在确认后帮你重启卡死的容器，形成"诊断 → 自愈"闭环。
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超时"
| 维度 | 评估 |
|---|---|
| 适合训练营类型 | DevOps/SRE 方向、Python 后端、AI 工程化训练营 |
| 教学时长 | 约 2.5 小时（0.5h 原理 + 1.5h 编码注册 + 0.5h 调试护栏） |
| 学员基础要求 | Python 基础 + 用过 docker，不要求懂 MCP 协议细节 |
| 可交付物 | 一个可复用的只读巡检 MCP + 团队级 .mcp.json 模板 |
| 拓展作业 | 加 restart_container 写工具并配 deny 护栏；改用 HTTP 传输部署成远程 MCP |
claude mcp add 与多服务器管理[!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 等场景全覆盖。