Kubernetes资源探索工具

一个智能的Kubernetes资源探索、关系映射和调试工具,支持CRD、AI驱动的洞察,并优化响应以供LLM使用。

  • by niradler
  • Python
  • 本地部署
  • 1 星标
  • 更新于
在 GitHub 查看 niradler/k8s-explorer-mcp

使用说明

🔍 K8s探索者MCP

用于Kubernetes资源探索、关系映射和调试的智能MCP服务器。了解CRD,提供AI驱动的见解,并优化LLM消费的响应。

✨ 特性

  • 🔍 智能资源发现 -查找pod使用的ConfigMaps/Secrets,检测Helm图表,跟踪操作员管理
  • 🌳 关系映射 -完整的父子链、服务路由、批量装载
  • 📦 13+CRD操作员 -Helm、ArgoCD、Airflow、Argo Workflows、Knative、FluxCD、Istio、证书管理器、Tekton、Spark、KEDA、Velero、Prometheus+AI驱动的未知CRD回退
  • ⚡ 智能缓存 -4层缓存,命中率超过80%,可快速重复查询
  • 🎯 响应滤波 -针对LLM消耗优化了70-90%的较小响应
  • 🔒 权限感知 -通过有用的解释适应RBAC约束
  • 🤖 AI驱动的洞察 -使用FastMCP采样的自然语言解释
  • 📝 内置提示 -预配置调试工作流提示
  • 🔄 多集群支持 -在多个K8s上下文之间无缝切换

🚀 快速开始

先决条件

  • Python 3.10或更高版本
  • kubectl配置了对Kubernetes集群的访问权限
  • Git

安装

# Install from PyPI (recommended)
uv pip install k8s-explorer-mcp

# Or install from source
git clone https://github.com/nirwo/k8s-explorer-mcp.git
cd k8s-explorer-mcp
uv pip install -e ".[dev]"

MCP配置

选项1:使用uvx(推荐-无需安装)

添加到光标MCP配置(~/.cursor/mcp.json 或 .cursor/mcp.json 在您的项目中):

{
  "mcpServers": {
    "k8s-explorer-mcp": {
      "command": "uvx",
      "args": [
        "--no-cache",
        "k8s-explorer-mcp"
      ]
    }
  }
}

此方法自动从PyPI下载并运行最新版本,无需手动安装。

选项2:地方发展设置

对于本地开发或测试未发布的更改:

{
  "mcpServers": {
    "k8s-explorer": {
      "command": "uv",
      "args": [
        "--directory",
        "/path/to/k8s-explorer-mcp",
        "run",
        "server.py"
      ],
      "env": {
        "PYTHONUNBUFFERED": "1"
      }
    }
  }
}

替换 /path/to/k8s-explorer-mcp 根据您的实际项目路径。

作为Python库

from k8s_explorer import K8sClient, K8sCache, RelationshipDiscovery
import asyncio

async def main():
    cache = K8sCache(resource_ttl=60, max_size=2000)
    client = K8sClient(cache=cache)
    discovery = RelationshipDiscovery(client)
    
    # Get a resource
    pods = await client.list_resources(
        kind="Pod",
        namespace="default",
        label_selector="app=nginx"
    )
    
    # Discover relationships
    if pods:
        relationships = await discovery.discover_relationships(pods[0])
        for rel in relationships:
            print(f"{rel.relationship_type}: {rel.target.kind}/{rel.target.name}")
    
    # Build resource tree
    from k8s_explorer.models import ResourceIdentifier
    resource_id = ResourceIdentifier(kind="Deployment", name="nginx", namespace="default")
    tree = await discovery.build_resource_tree(resource_id, max_depth=3)
    
    stats = client.get_cache_stats()
    print(f"Cache hit rate: {stats['hit_rate_percent']}%")

asyncio.run(main())

作为MCP服务器

# Start the server
uv run server.py

# Or with make
make run

可用工具 (9个精简工具):

核心业务(4)

  • list_contexts() -列出可用上下文和可访问的命名空间
  • list_resources(kind, namespace, labels, all_namespaces) -通用列出任何资源类型
  • get_resource(kind, name, namespace) -获取特定资源(Pod的智能匹配)
  • kubectl(args, namespace) -执行kubectl命令以获得灵活性

发现(1)

  • discover_resource(kind, name, namespace, depth="complete") -一个工具满足所有发现需求
    • deputy=“关系”:快速连接列表(所有者、子用户、卷、CRD)
    • 深度=“树”:显示资源层次结构的层次树结构
    • 深度=“完成”:调试的完整上下文(默认)-包括管理信息、解释
    • 用清晰的深度参数替换3个单独的工具

日志(1)

  • get_pod_logs(name, namespace, container, previous, tail, timestamps) -获取pod日志
    • 针对LLM消耗进行了优化
    • 自动处理多容器吊舱
    • 显示截断信息和可用容器

变更跟踪(2)

  • get_resource_changes(kind, name, namespace, max_versions) -变更时间表

    • 显示版本之间的更改
    • LLM使用max_versions控制深度
  • compare_resource_versions(kind, name, namespace, from_revision, to_revision) -版本比较

    • 详细的逐字段比较

图形分析(1)

  • build_resource_graph(namespace, kind, name, depth, include_rbac, include_network, include_crds) -构建完整的资源图
    • 两种模式:特定资源或完整命名空间
    • 带缓存的增量图构建
    • RBAC、网络策略和CRD关系支持

内置提示(1)

  • debug_failing_pod(pod_name, namespace) -通过指导调查步骤完成故障吊舱的调试工作流程

所有工具和提示都具有权限意识 并将:

  • 根据RBAC权限调整响应
  • 在访问受限时包含权限通知
  • 为缺失的权限提供明确的指导

🎯 我们能发现什么?

✅ K8s运营所需的一切LLM

配置映射和秘密:自动查找pod使用的所有配置资源(卷挂载、环境变量、投影卷)

Helm图表:检测哪个Helm chart创建了任何资源(发布名称、图表版本、所有托管资源)

操作员(13+):标识由Helm、ArgoCD、Argo工作流、Airflow、Knative、FluxCD、Istio、证书管理器、Tekton、Spark、KEDA、Velero、Prometheus+AI驱动的未知CRD回退管理的资源

完整的关系:父子链、服务路由、卷依赖关系、标签选择器、操作员管理

📚 文档

文档 描述
特工.md 全面的代理商指南
贡献.md 如何做出贡献
示例/ 使用示例

📦 项目结构

k8s-explorer-mcp/
├── server.py                 # MCP server
├── k8s_explorer/             # Main package
│   ├── client.py             # Kubernetes client wrapper
│   ├── cache.py              # Multi-layer caching
│   ├── config.py             # Configuration system
│   ├── models.py             # Data models
│   └── operators/            # CRD operator support
├── examples/                 # Usage examples
├── tests/                    # Test suite
└── pyproject.toml            # Project metadata

🎯 用例

🧠 智能吊舱匹配(新)

# Pod was recreated with different suffix? No problem!
get_resource("Pod", "myapp-deployment-abc123-old999", "default")

# Automatically finds and returns:
# {
#   "name": "myapp-deployment-def456-xyz789",
#   "match_info": {
#     "fuzzy_match_used": true,
#     "original_name": "myapp-deployment-abc123-old999",
#     "matched_name": "myapp-deployment-def456-xyz789",
#     "similarity_score": 1.0,
#     "match_reason": "exact_base_match",
#     "explanation": "Pod 'myapp-deployment-abc123-old999' not found, but found 
#                     'myapp-deployment-def456-xyz789' with same base name 
#                     'myapp-deployment'. This is likely a newer instance."
#   }
# }

# Search for pods by pattern (fuzzy matching built-in)
get_resource("Pod", "cronjob-backup", "default")

# Automatically finds similar pods with similarity scores
# Handles: Deployments, StatefulSets, Jobs, CronJobs suffixes

🐛 调试变得容易

# Get complete context for debugging (smart matching included)
discover_resource("Pod", "my-app-xyz", "production", depth="complete")

# Returns:
# - ConfigMaps it needs (and if they exist)
# - Secrets it uses (with mount details)
# - Parent Deployment/ReplicaSet
# - Helm chart managing it
# - Complete failure context with explanations
# - Match info if pod name was fuzzy matched

# Get pod logs (with automatic container detection)
get_pod_logs("my-app-xyz", "production", tail=200)

# Returns:
# - Logs from the pod (auto-detects single container)
# - Pod status and container list
# - Truncation info
# - Match info if fuzzy matching was used

📊 变更跟踪和调查(新)

# What changed in the last deployment?
get_resource_changes("Deployment", "nginx-deployment", "production", max_versions=3)

# Returns:
# {
#   "latest_changes": {
#     "from_revision": "5",
#     "to_revision": "6",
#     "summary": "2 field(s) modified",
#     "changes": [
#       {
#         "field": "spec.replicas",
#         "change_type": "modified",
#         "old_value": "3",
#         "new_value": "5",
#         "delta": 2,
#         "percent_change": 66.67
#       },
#       {
#         "field": "spec.template.spec.containers[0].image",
#         "change_type": "modified",
#         "old_value": "nginx:1.21",
#         "new_value": "nginx:1.22"
#       }
#     ]
#   },
#   "timeline": [...],  # History of all changes
#   "note": "Use max_versions to control how far back to look"
# }

# Compare specific versions
compare_resource_versions("Deployment", "nginx", "prod", from_revision=3, to_revision=5)

# Get full change history (defaults to last 5 versions)
get_resource_changes("Deployment", "nginx", "prod")
# Returns: Timeline of changes with diffs

🔍 影响分析

# What will break if I delete this ConfigMap?
discover_resource("ConfigMap", "app-config", "prod", depth="tree")

# Shows:
# - All Deployments using it
# - All ReplicaSets affected
# - All Pods that will restart

# Quick check of dependencies
discover_resource("Secret", "db-password", "prod", depth="relationships")
# Fast list of all resources using this secret

🚀 操作员调试

# Debug Airflow DAG - find related resources
list_resources("Pod", "airflow", labels={"dag_id": "etl-pipeline"})

# Debug Argo Workflow - full context
discover_resource("Workflow", "data-processing", "workflows", depth="complete")

# Debug Helm release - shows chart, version, all managed resources
discover_resource("Deployment", "nginx", "default", depth="complete")
# Returns: Helm release name, chart version, all related resources

🤖 LLM动力操作

法学硕士现在可以理解:

  • “显示nginx Helm chart创建的所有Pod”
  • “此部署使用什么ConfigMgr?”
  • “为什么我的气流任务失败了?”
  • “如果我更新这个秘密,会有什么结果?”

所有与 一个工具调用 和 完整上下文!

🧪 运行测试

# Run all tests
make test

# Run with coverage report
make test-cov

# Run specific test file
pytest tests/test_cache.py

🤝 贡献

我们欢迎捐款!请看 贡献.md 作为指导方针。

开发环境设置

# Install with dev dependencies
make install-dev

# Format code
make format

# Lint code
make lint

# Run all checks
make check

📝 许可证

MIT许可证-请参阅 许可证 文件以获取详细信息。


由以下材料制成❤️ Kubernetes社区