K8s MCP Server

K8s MCP 服务

增强型 Kubernetes 访问，提供支持无头服务的 AI 友好型端口转发管理。

✅ 实现已完成 — 所有核心功能均已实现。

功能

• 受 kubectl 启发的输出格式：支持多种输出格式（table、wide、json、yaml、custom-columns、jsonpath），高效利用令牌
• 资源类型发现：通过自然语言查询，根据关键词查找自定义资源定义和内置资源
• 基于 URL 的配置：通过 GitHub Gists 或本地文件共享端口转发配置
• 无头服务支持：通过标签选择器直接定位 Pod
• 针对 AI 优化：简洁的服务/目标结构，便于智能模糊匹配
• 多格式支持：YAML、JSON、TOML 配置文件
• 类似 kubectl 的行为：选择第一个就绪的 Pod，与 kubectl port-forward 相同
• 会话管理：在内存中跟踪端口转发，支持列出和拆除会话
• 无需参数即可发现：便于 AI Agent 发现集群、别名和服务

快速开始

1. 安装依赖

pip install -e .

2. 设置环境变量

# Required: Kubernetes cluster configurations
export KUBECONFIG_URLS="dev=file:///path/to/dev.yaml prod=file:///path/to/prod.yaml"

# Optional: Cluster aliases
export KUBECONFIG_ALIASES="development=dev production=prod"

# Required: Port-forward configuration URL
export PORT_FORWARD_CONFIG_URL="file://$(pwd)/example-port-forward-config.yaml"
# Or use a GitHub Gist:
# export PORT_FORWARD_CONFIG_URL="https://gist.githubusercontent.com/user/abc123/raw/port-forward.yaml"

3. 运行服务

python main.py

使用示例

AI Agent 交互

User: "Setup port-forward to the FRET Druid in development"
AI Agent:
1. Calls discover_port_forward_services() to understand available services
2. Maps "FRET Druid in development" → service="druid", target="fret-dev"  
3. Calls setup_port_forward_to_target("druid", "fret-dev")
4. Returns: "✅ Connected to fret-dev Druid: http://localhost:8082 → pod/druid-router-0"

直接调用工具

• connect_to_service_targets("druid", ["fret-dev"]) — 连接到指定目标
• connect_to_service_targets("druid", []) — 连接到所有 Druid 环境
• list_resources("prod", "pods", labels={"app": "nginx"}) — 列出带标签的 Pod
• get_resource("dev", "deployment", "my-app", "default") — 获取指定资源
• discover_resource_kinds("dev", ["druid", "ingestion"]) — 根据关键词查找资源类型
• list_resources("dev", "pods", output_format="custom-columns", custom_columns="NAME:.metadata.name,IP:.status.podIP") — 自定义输出
• list_port_forwards() — 显示活跃的端口转发会话
• teardown_port_forward("forward-id") — 停止端口转发会话
• 支持 kubectl 命名变体："pods"、"pod"、"po" 均可使用

输出格式示例

高效利用令牌的资源列表

# Default table format (kubectl-like)
list_resources(cluster="dev", kind="pods", namespace="fret-dev")
# Returns clean table: NAME | READY | STATUS | RESTARTS | AGE

自定义列以获取指定数据

# Extract specific fields with JSONPath
list_resources(
    cluster="dev", 
    kind="services",
    output_format="custom-columns", 
    custom_columns="NAME:.metadata.name,TYPE:.spec.type,PORTS:.spec.ports[*].port"
)

资源类型发现

# Find resource kinds by keywords
discover_resource_kinds(
    cluster="dev",
    keywords=["druid", "ingestion"],
    match_mode="OR"
)
# Returns: DruidIngestion resource kind info with all naming variants

资源命名约定

K8s MCP 服务遵循 kubectl 命名约定：

支持的资源类型

• 复数形式：pods、services、deployments
• 单数形式：pod、service、deployment
• 缩写形式：po、svc、deploy、sts（表示 StatefulSet）
• 不区分大小写：Pod、POD、pod 均可使用

发现与搜索

• discover_* 工具用于查找可用的资源类型/类型
• search_* 和 list_* 工具用于查找具体实例

工作流示例

# 1. Discover what kinds are available
discover_resource_kinds(cluster="dev", keywords=["druid"])
# Returns: DruidIngestion kind with variants ["druidingestions", "druidingestion"]

# 2. List instances using any variant
list_resources(cluster="dev", kind="druidingestions")
# or
list_resources(cluster="dev", kind="DruidIngestion")

配置

端口转发配置格式

druid:
  fret-dev:
    k8s_cluster: "dev-us-east"
    namespace: "druid"
    labels:
      app.kubernetes.io/component: "router"
    local_port: 8082
    remote_port: 8888

grafana:
  dev:
    k8s_cluster: "dev-us-east"
    namespace: "monitoring"
    labels:
      app.kubernetes.io/name: "grafana"
    local_port: 3000
    remote_port: 3000

常见工作流

以下工作流示例展示了 AI Agent 如何组合多个工具来完成复杂的 Kubernetes 任务：

1. 集群概览工作流

User: "Show me the status of all my clusters"
AI Agent Flow:
1. discover_clusters() → Get available clusters ["dev", "staging", "prod"]
2. For each cluster:
   - cluster_health() → Check API server connectivity
   - list_resources("cluster", "nodes") → Get node count and status
   - list_resources("cluster", "namespaces") → Get namespace overview
3. Summarize: "3 clusters online, 12 total nodes, 8 namespaces"

2. 应用调试工作流

User: "My app isn't working in staging"
AI Agent Flow:
1. list_resources("staging", "pods", labels={"app": "user-app"}) → Find app pods
2. For each pod:
   - describe_resource("staging", "pod", pod_name) → Get detailed status
   - get_logs("staging", pod_name, lines=50) → Check recent logs
   - list_resources("staging", "events", field_selector=f"involvedObject.name={pod_name}") → Get events
3. Analyze patterns: "Found 2 pods crash-looping due to missing config map 'app-config'"

3. 安全部署更新工作流

User: "Update nginx deployment to version 1.21"
AI Agent Flow:
1. get_resource("prod", "deployment", "nginx") → Get current deployment
2. Modify image version in manifest
3. Present proposed changes: "Will update image from nginx:1.20 to nginx:1.21"
4. User confirms: "Apply it"
5. update_resource("prod", "deployment", "nginx", new_manifest) → Apply update
6. Monitor rollout status with deployment_status()

4. 资源清理工作流

User: "Clean up all test resources from last week"
AI Agent Flow:
1. list_resources("dev", "all", labels={"purpose": "test", "created": "2024-01-15"}) → Find test resources
2. Group by type: "Found 5 pods, 3 services, 2 configmaps"
3. For each resource:
   - Show resource details and age
4. Present deletion plan with resource details
5. User confirms: "Yes, delete them"
6. Execute deletions with proper cascade handling

5. 增强型端口转发工作流

User: "Setup port-forward to the FRET Druid in development"
AI Agent Flow:
1. discover_port_forward_services() → Get available services/targets
2. AI reasoning: "FRET Druid in development" → service="druid", target="fret-dev"
3. connect_to_service_targets("druid", ["fret-dev"]) → Setup to specific target
4. Return: "✅ Connected to fret-dev Druid: http://localhost:8082 → pod/druid-router-0"

6. 性能故障排查工作流

User: "Why is my app slow in production?"
AI Agent Flow:
1. list_resources("prod", "pods", labels={"app": "slow-app"}) → Find app pods
2. For each pod:
   - describe_resource() → Check resource limits/requests
   - get_logs(since="1h") → Check for performance-related logs
   - execute_in_pod(["top", "-b", "-n", "1"]) → Get resource usage
3. list_resources("prod", "nodes") → Check node capacity
4. query_by_labels("prod", "hpa", "app=slow-app") → Check auto-scaling
5. Analyze: "App is CPU-throttled, increase CPU requests from 100m to 250m"

7. 安全审计工作流

User: "Audit RBAC permissions for the finance team"
AI Agent Flow:
1. list_resources("prod", "rolebindings") → Get all role bindings
2. list_resources("prod", "clusterrolebindings") → Get cluster-wide bindings
3. Filter by subjects containing "finance"
4. For each binding:
   - get_resource("prod", "role", role_name) → Get role permissions
5. Summarize: "Finance team has read access to secrets in finance namespace"

8. 灾难恢复工作流

User: "Restore the database from backup"
AI Agent Flow:
1. list_resources("prod", "pods", labels={"app": "database"}) → Find DB pods
2. scale_resource("prod", "deployment", "database", replicas=0) → Scale down
3. Wait for pods to terminate
4. apply_manifest("prod", restore_job_manifest) → Apply restore job
5. Monitor job status until completion
6. scale_resource("prod", "deployment", "database", replicas=1) → Scale back up
7. Verify database connectivity

AI Agent 最佳实践

1. 始终预览更改：修改前使用 diff/dry-run
2. 提供上下文：报告问题时附上相关信息
3. 批量操作：将相关操作分组以提高效率
4. 处理分页：留意大型结果集
5. 遵守集群边界：切勿假设操作会跨越集群

工具安全分类

所有 MCP 工具均按其在 AI Agent 操作中的安全级别进行分类：

[READ-ONLY] 工具 — 可安全自主使用

这些工具只读取数据，不会修改集群或本地状态：

资源操作：

• get_resource() - 获取特定资源的详细信息
• get_resource_yaml() - 以 YAML 格式获取资源
• list_resources() - 按格式选项列出资源
• list_resources_yaml() - 以 YAML 格式列出资源
• query_by_labels() - 基于标签进行高级查询
• query_by_labels_yaml() - 以 YAML 格式进行标签查询

监控与分析：

• troubleshoot_pod() - 全面调试 Pod
• troubleshoot_pod_yaml() - 以 YAML 格式调试 Pod
• cluster_health() - 整体评估集群健康状况
• deployment_status() - 分析 Deployment 的滚动更新状态
• deployment_status_yaml() - 以 YAML 格式分析 Deployment
• resource_usage() - 分析资源消耗
• get_logs() - 获取 Pod 日志
• list_port_forwards() - 列出活跃的端口转发会话

发现：

• discover_resource_kinds() - 按关键词查找资源类型
• discover_clusters() - 列出可用集群
• discover_cluster_aliases() - 列出集群别名
• discover_port_forward_services() - 列出端口转发配置

[MODIFIES STATE] 工具 — 需要用户批准

这些工具会修改集群状态或本地网络配置：

资源管理：

• create_resource() - 创建新的 Kubernetes 资源
• update_resource() - 更新现有资源
• patch_resource() - 部分更新资源
• scale_resource() - 更改副本数量
• apply_manifest() - 应用 YAML 清单

网络操作：

• connect_to_service_targets() - 连接到已配置的服务环境
• setup_port_forward() - 设置通用端口转发
• teardown_port_forward() - 关闭端口转发会话

[DANGEROUS] 工具 — 需要用户明确确认

这些工具可能导致数据丢失或安全风险：

• delete_resource() - 永久删除资源（无法撤销）
• execute_in_pod() - 在 Pod 中运行任意命令（存在潜在安全风险）

AI Agent 安全指南

1. 自主操作：AI Agent 可以自由使用 [READ-ONLY] 工具
2. 状态更改：[MODIFIES STATE] 工具应要求用户批准
3. 高风险操作：[DANGEROUS] 工具必须获得用户明确确认
4. 错误处理：所有工具都会返回统一的错误格式，以便安全解析

架构

k8s-mcp/
├── main.py                 # MCP server entry point with tool definitions
├── config/                 # Configuration management  
│   ├── loader.py          # URL-based config loading
│   ├── manager.py         # Cluster and port-forward session management
│   └── config.py          # Environment validation and configuration
├── tools/                  # Modular tool implementations
│   ├── resources.py       # Common resource operation patterns
│   ├── formatting.py      # All resource formatting (table, YAML, JSON)
│   ├── columns.py         # Default column definitions for resources
│   ├── discovery.py       # Resource kind discovery and CRD enumeration
│   └── port_forward.py    # Port-forward specific operations
└── utils/                  # Shared utilities
    ├── helpers.py         # Common helper functions
    └── jsonpath.py        # JSONPath processing for custom columns

代码组织原则

• 单一职责：每个模块都有明确且专注的用途
• 遵循 DRY 原则：各文件之间不重复实现
• 关注点分离：客户端管理、格式化和业务逻辑彼此分离
• 模块化设计：工具可以独立导入和复用