使用说明

K8s MCP服务器

通过支持无头服务的AI友好端口转发管理增强Kubernetes访问。

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

特性

kubectl启发的输出格式：多种输出格式（表、宽、json、yaml、自定义列、jsonpath），可实现高效的令牌使用
资源种类发现：按关键字查找自定义资源定义和内置资源的自然语言查询
基于URL的配置：通过GitHub Gists或本地文件共享端口转发配置
无头服务支持：通过标签选择器直接定位吊舱
AI优化：智能模糊匹配的清洁服务/目标结构
多格式支持：YAML、JSON、TOML配置文件
类似kubectl的行为：选择第一个就绪吊舱，与 kubectl port-forward
会话管理：具有列表和拆卸功能的内存端口前向跟踪
无参数发现：为AI代理提供简单的集群、别名和服务发现

快速开始

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代理交互

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", []) -连接到所有德鲁伊环境
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 所有工作

发现vs搜索

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代理如何组合多种工具来完成复杂的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代理的最佳实践

始终预览更改：修改前使用差分/试运行
提供上下文：在报告问题时包括相关信息
批量操作：提高效率的集团相关业务
处理分页：注意大型结果集
尊重集群边界：切勿假设跨集群操作

工具安全分类

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

[只读]工具-可安全自主使用

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

资源操作：

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

监测和分析：

troubleshoot_pod() -吊舱综合调试
troubleshoot_pod_yaml() -Pod调试为YAML
cluster_health() -总体集群健康评估
deployment_status() -部署部署分析
deployment_status_yaml() -部署分析为YAML
resource_usage() -资源消耗分析
get_logs() -检索吊舱日志
list_port_forwards() -列出活动端口转发会话

发现：

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

[修改状态]工具-需要用户批准

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

资源管理：

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

网络操作：

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

[危险]工具-需要用户明确确认

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

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

AI代理安全指南

自主操作：AI代理可以自由使用[只读]工具
状态变化：[修改状态]工具应需要用户批准
高风险操作：[危险]工具必须有明确的用户确认
错误处理：所有工具都返回一致的错误格式以进行安全解析

建筑

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

代码组织原则

单一责任：每个模块都有明确、重点突出的目的
干燥合规性：文件之间没有重复的实现
关注点分离：客户端管理、格式化和业务逻辑是分开的
模块化设计：工具可以独立导入和重复使用