一个基于 FastMCP 框架构建的 Kubernetes 管理服务器,提供完整的 Kubernetes 集群管理功能。
- 完整的 Kubernetes 管理: 支持 kubectl 的所有核心操作
- Helm 集成: 提供 Helm Chart 的安装、升级、卸载功能
- 多种传输方式: 支持 HTTP 和 STDIO 两种传输协议
- 灵活的配置: 支持多种 kubeconfig 配置方式
- 安全性: 自动屏蔽 Secret 敏感数据
- 故障诊断: 内置 Kubernetes 故障诊断助手
- 健康检查: 提供服务健康状态监控
MCP/
├── core/
│ └── kubernetes_mcp_core.py # 核心模块:包含所有工具函数实现
├── httpserver.py # HTTP 传输版本的 MCP 服务器
├── stdio.py # STDIO 传输版本的 MCP 服务器
├── simple_mcp_test.py # 简单的测试客户端
├── pyproject.toml # 项目配置和依赖管理
├── VERSION # 版本信息
└── README.md # 项目文档
core/kubernetes_mcp_core.py: 核心模块,包含所有 Kubernetes 和 Helm 操作的实现httpserver.py: HTTP 传输模式的服务器入口stdio.py: STDIO 传输模式的服务器入口(现在支持完整功能)
- Python 3.10+
- kubectl (必需)
- helm (可选,用于 Helm 功能)
- 有效的 Kubernetes 集群访问权限
项目使用 uv 作为包管理器:
# 安装依赖
uv sync
# 开发环境依赖
uv sync --dev服务器支持多种 kubeconfig 配置方式,按优先级排序:
-
YAML 环境变量:
export KUBECONFIG_YAML="$(cat ~/.kube/config)"
-
JSON 环境变量:
export KUBECONFIG_JSON='{"apiVersion":"v1","kind":"Config",...}'
-
最小配置:
export K8S_SERVER="https://your-k8s-api-server" export K8S_TOKEN="your-service-account-token" export K8S_SKIP_TLS_VERIFY="true" # 可选
-
标准路径:
export KUBECONFIG="/path/to/your/kubeconfig" # 或使用默认路径 ~/.kube/config
# 设置默认上下文
export K8S_CONTEXT="your-context-name"
# 设置默认命名空间
export K8S_NAMESPACE="your-namespace"
# 控制 Secret 数据屏蔽(默认启用)
export MASK_SECRETS="true"适用于需要 HTTP API 访问的场景:
uv run httpserver.py服务器将在 http://127.0.0.1:6000 启动。
适用于 MCP 客户端(如 Cherry Studio):
uv run stdio.pyping: 验证与 Kubernetes 集群的连接kubectl_get: 获取或列出 Kubernetes 资源kubectl_describe: 描述资源的详细信息kubectl_apply: 应用 YAML 清单kubectl_delete: 删除 Kubernetes 资源kubectl_logs: 获取资源日志kubectl_context: 管理 Kubernetes 上下文kubectl_scale: 扩缩容资源kubectl_patch: 更新资源字段kubectl_rollout: 管理滚动更新kubectl_exec: 在 Pod 中执行命令
helm_install: 安装 Helm Charthelm_upgrade: 升级 Helm Releasehelm_uninstall: 卸载 Helm Releasehelm_list: 列出 Helm Releases
health_check: 健康检查端点
k8s://cluster/info: 获取集群信息k8s://contexts: 获取可用上下文
k8s_diagnose: Kubernetes 故障诊断助手
使用提供的测试客户端验证服务器功能:
# 确保 HTTP 服务器正在运行
uv run simple_mcp_test.py项目采用模块化设计,将核心功能与传输层分离:
- 核心模块 (
core/kubernetes_mcp_core.py): 包含所有业务逻辑和工具实现 - 传输层 (
httpserver.py,stdio.py): 负责不同的通信协议
- 代码复用: 两种传输模式共享相同的核心功能
- 易于维护: 业务逻辑集中在核心模块中
- 功能一致: STDIO 和 HTTP 版本提供完全相同的功能
- 扩展性: 可以轻松添加新的传输模式
- 清晰分层: 核心逻辑与传输协议完全分离
- 单一职责: 每个模块都有明确的职责边界
- 依赖倒置: 传输层依赖于核心模块,而非相反
- 开放封闭: 对扩展开放(新传输模式),对修改封闭(核心逻辑)
from fastmcp import Client
import asyncio
async def example():
async with Client("http://127.0.0.1:6000/mcp") as client:
# 检查连接
result = await client.call_tool("ping")
print(result)
# 获取所有 Pod
pods = await client.call_tool("kubectl_get", {
"resource_type": "pods",
"all_namespaces": True
})
print(pods)
# 获取特定 Deployment 的日志
logs = await client.call_tool("kubectl_logs", {
"resource_type": "deployment",
"name": "my-app",
"namespace": "default",
"tail": 100
})
print(logs)
asyncio.run(example())# 安装 Chart
await client.call_tool("helm_install", {
"name": "my-release",
"chart": "stable/nginx",
"namespace": "default",
"values": {
"replicaCount": 3,
"service": {
"type": "LoadBalancer"
}
}
})- Secret 数据屏蔽: 自动屏蔽 Kubernetes Secret 中的敏感数据
- 超时控制: 所有 kubectl 和 helm 命令都有超时限制
- 错误处理: 完善的错误处理和日志记录
-
kubectl 未找到:
# 确保 kubectl 已安装并在 PATH 中 kubectl version --client -
连接失败:
# 检查 kubeconfig 配置 kubectl cluster-info -
权限问题:
# 检查当前用户权限 kubectl auth can-i --list
服务器启动时会显示配置信息:
启动Kubernetes MCP服务器...
配置信息:
- kubeconfig路径: /home/user/.kube/config
- 默认命名空间: default
- 上下文: my-context
欢迎提交 Issue 和 Pull Request!
本项目采用 MIT 许可证。