fix readme

saberrey · saberrey · commit e2efd7798299 · 2025-10-20T20:10:34.000+08:00
diff --git a/README.md b/README.md
@@ -34,6 +34,7 @@ https://github.com/user-attachments/assets/9e48cac3-0af1-424c-9f16-3862d047cc68
 - 集群资源运维任务查询 (Later)
 
 **Kubernetes 原生操作** (`ack_kubectl`)
+- 【注意⚠️】`ack_kubectl`会在`~/.kube`下生成和管理临时 kubeconfig，并在工具内部自动切换至正确上下文；它不会修改全局默认上下文或用户的`KUBECONFIG`。为避免与其他入口（如 shell 中的 `kubectl`）产生集群不一致，请在访问 ACK 集群时仅使用该工具，不要从外部直接执行 `kubectl`。
 - 执行 `kubectl` 类操作（读写权限可控）
 - 获取日志、事件，资源的增删改查
 - 支持所有标准 Kubernetes API
diff --git a/src/kubectl_handler.py b/src/kubectl_handler.py
@@ -4,10 +4,225 @@
 import os
 import subprocess
 from typing import Dict, Optional
+from cachetools import TTLCache
 from loguru import logger
+from ack_cluster_handler import parse_master_url
 from models import KubectlOutput
 
 
+class KubectlContextManager(TTLCache):
+    """基于 TTL+LRU 缓存的 kubeconfig 文件管理器"""
+
+    def __init__(self, ttl_minutes: int = 60):
+        """初始化上下文管理器
+
+        Args:
+            ttl_minutes: kubeconfig有效期（分钟），默认60分钟
+        """
+        # 初始化 TTL+LRU 缓存
+        super().__init__(maxsize=50, ttl=ttl_minutes * 60)  # TTL 以秒为单位，提前5min
+
+        self._cs_client = None  # CS客户端实例
+
+        # 使用 .kube 目录存储 kubeconfig 文件
+        self._kube_dir = os.path.expanduser("~/.kube")
+        os.makedirs(self._kube_dir, exist_ok=True)
+
+        self._setup_cleanup_handlers()
+
+    def _setup_cleanup_handlers(self):
+        """设置清理处理器"""
+        import atexit
+        import signal
+
+        def cleanup_contexts():
+            """清理所有上下文"""
+            try:
+                context_manager = get_context_manager()
+                if context_manager:
+                    context_manager.cleanup()
+                else:
+                    self.cleanup_all_mcp_files()
+            except Exception as e:
+                logger.error(f"Cleanup failed: {e}")
+                raise e
+
+        def signal_handler(signum, frame):
+            """信号处理器"""
+            cleanup_contexts()
+            exit(0)
+
+        atexit.register(cleanup_contexts)
+        signal.signal(signal.SIGINT, signal_handler)
+        signal.signal(signal.SIGTERM, signal_handler)
+
+    def cleanup_all_mcp_files(self):
+        """类方法：清理所有MCP创建的kubeconfig文件（安全清理）"""
+        try:
+            kube_dir = os.path.expanduser("~/.kube")
+            if not os.path.exists(kube_dir):
+                return
+
+            removed_count = 0
+            for filename in os.listdir(kube_dir):
+                if filename.startswith("mcp-kubeconfig-") and filename.endswith(".yaml"):
+                    file_path = os.path.join(kube_dir, filename)
+                    try:
+                        os.remove(file_path)
+                        removed_count += 1
+                    except Exception:
+                        pass
+
+            if removed_count > 0:
+                print(f"Cleaned up {removed_count} MCP kubeconfig files")
+        except Exception:
+            pass
+
+    def _get_or_create_kubeconfig_file(self, cluster_id: str) -> str:
+        """获取或创建集群的 kubeconfig 文件
+
+        Args:
+            cluster_id: 集群ID
+
+        Returns:
+            kubeconfig 文件路径
+        """
+        # 检查缓存中是否已存在
+        if cluster_id in self:
+            logger.debug(f"Found cached kubeconfig for cluster {cluster_id}")
+            return self[cluster_id]
+
+        # 创建新的 kubeconfig 文件
+        kubeconfig_content = self._get_kubeconfig_from_ack(cluster_id, int(self.ttl / 60))  # 转换为分钟
+        if not kubeconfig_content:
+            raise ValueError(f"Failed to get kubeconfig for cluster {cluster_id}")
+
+        # 创建 kubeconfig 文件
+        kubeconfig_path = os.path.join(self._kube_dir, f"mcp-kubeconfig-{cluster_id}.yaml")
+
+        # 确保目录存在
+        os.makedirs(os.path.dirname(kubeconfig_path), exist_ok=True)
+
+        with open(kubeconfig_path, 'w') as f:
+            f.write(kubeconfig_content)
+
+        # 添加到缓存
+        self[cluster_id] = kubeconfig_path
+        return kubeconfig_path
+
+    def popitem(self):
+        """重写 popitem 方法，在驱逐缓存项时清理 kubeconfig 文件"""
+        key, path = super().popitem()
+        # 删除 kubeconfig 文件
+        if path and os.path.exists(path):
+            try:
+                os.remove(path)
+                logger.debug(f"Removed cached kubeconfig file: {path}")
+            except Exception as e:
+                logger.warning(f"Failed to remove cached kubeconfig file {path}: {e}")
+
+        return key, path
+
+    def cleanup(self):
+        """清理资源，删除所有 MCP 创建的 kubeconfig 文件和缓存"""
+        removed_count = 0
+        for key, path in list(self.items()):
+            if path and os.path.exists(path):
+                try:
+                    os.remove(path)
+                    removed_count += 1
+                except Exception:
+                    pass
+        self.clear()
+        print(f"Cleaned up {removed_count} kubeconfig files")
+
+    def set_cs_client(self, cs_client):
+        """设置CS客户端
+
+        Args:
+            cs_client: CS客户端实例
+        """
+        self._cs_client = cs_client
+
+    def _get_cs_client(self):
+        """获取CS客户端"""
+        if not self._cs_client:
+            raise ValueError("CS client not set")
+        return self._cs_client
+
+    def _get_kubeconfig_from_ack(self, cluster_id: str, ttl_minutes: int = 60) -> Optional[str]:
+        """通过ACK API获取kubeconfig配置
+
+        Args:
+            cluster_id: 集群ID
+            ttl_minutes: kubeconfig有效期（分钟），默认60分钟
+        """
+        try:
+            # 获取CS客户端
+            cs_client = self._get_cs_client()
+            from alibabacloud_cs20151215 import models as cs_models
+
+            # 先检查集群详情，确认是否有公网端点
+            detail_response = cs_client.describe_cluster_detail(cluster_id)
+
+            if not detail_response or not detail_response.body:
+                raise ValueError(f"Failed to get cluster details for {cluster_id}")
+
+            cluster_info = detail_response.body
+            # 检查是否有公网API Server端点
+            master_url_str = getattr(cluster_info, 'master_url', '')
+            master_url = parse_master_url(master_url_str)
+            if not master_url["api_server_endpoint"]:
+                raise ValueError(f"Cluster {cluster_id} does not have public endpoint access, "
+                                 f"Please enable public endpoint access setting first.")
+
+            # 调用DescribeClusterUserKubeconfig API
+            request = cs_models.DescribeClusterUserKubeconfigRequest(
+                private_ip_address=False,  # 获取公网连接配置
+                temporary_duration_minutes=ttl_minutes,  # 使用传入的TTL
+            )
+
+            response = cs_client.describe_cluster_user_kubeconfig(cluster_id, request)
+
+            if response and response.body and response.body.config:
+                logger.info(f"Successfully fetched kubeconfig for cluster {cluster_id} (TTL: {ttl_minutes} minutes)")
+                return response.body.config
+            else:
+                logger.warning(f"No kubeconfig found for cluster {cluster_id}")
+                return None
+
+        except Exception as e:
+            logger.error(f"Failed to fetch kubeconfig for cluster {cluster_id}: {e}")
+            raise e
+
+    def get_kubeconfig_path(self, cluster_id: str) -> str:
+        """获取集群的 kubeconfig 文件路径
+
+        Args:
+            cluster_id: 集群ID
+
+        Returns:
+            kubeconfig 文件路径
+        """
+        return self._get_or_create_kubeconfig_file(cluster_id)
+
+
+# 全局上下文管理器实例
+_context_manager: Optional[KubectlContextManager] = None
+
+
+def get_context_manager(ttl_minutes: int = 60) -> KubectlContextManager:
+    """获取全局上下文管理器实例
+
+    Args:
+        ttl_minutes: kubeconfig有效期（分钟），默认60分钟
+    """
+    global _context_manager
+    if _context_manager is None:
+        _context_manager = KubectlContextManager(ttl_minutes=ttl_minutes)
+    return _context_manager
+
+
 class KubectlHandler:
     """Handler for running kubectl commands via a FastMCP tool."""
 
@@ -31,20 +246,29 @@ def __init__(self, server: FastMCP, settings: Optional[Dict[str, Any]] = None):
         
         self._register_tools()
 
-    def _get_kubeconfig_path(self) -> str:
-        """获取本机kubeconfig文件路径"""
-        # 优先使用KUBECONFIG环境变量
-        kubeconfig_path = os.environ.get('KUBECONFIG')
-        if kubeconfig_path and os.path.exists(kubeconfig_path):
-            return kubeconfig_path
-        
-        # 使用默认路径
-        default_path = os.path.expanduser("~/.kube/config")
-        if os.path.exists(default_path):
-            return default_path
-        
-        # 如果都不存在，返回默认路径让kubectl处理
-        return default_path
+    def _setup_cs_client(self, ctx: Context):
+        """设置CS客户端（仅在需要时）"""
+        try:
+            # 检查是否已经设置过
+            if hasattr(get_context_manager(), '_cs_client') and get_context_manager()._cs_client:
+                return
+
+            lifespan_context = ctx.request_context.lifespan_context
+            if isinstance(lifespan_context, dict):
+                providers = lifespan_context.get("providers", {})
+            else:
+                providers = getattr(lifespan_context, "providers", {})
+
+            cs_client_factory = providers.get("cs_client_factory")
+            if cs_client_factory:
+                # 传入统一签名所需的 config
+                config = lifespan_context.get("config", {}) if isinstance(lifespan_context, dict) else {}
+                get_context_manager().set_cs_client(cs_client_factory("CENTER", config))
+                logger.debug("CS client factory set successfully")
+            else:
+                logger.warning("cs_client not available in lifespan context")
+        except Exception as e:
+            logger.error(f"Failed to setup CS client: {e}")
 
     def is_write_command(self, command: str) -> tuple[bool, Optional[str]]:
         """检查是否为可写命令
@@ -280,9 +504,17 @@ async def ack_kubectl(
 user: I need to execute a command in the pod
 assistant: exec my-pod -- /bin/sh -c "your command here"""
                 ),
+                cluster_id: str = Field(
+                    ..., description="The ID of the Kubernetes cluster to query. If specified, will auto find/create "
+                                     "and switch to appropriate context. If you are not sure of cluster id, "
+                                     "please use the list_clusters tool to get it first."
+                ),
         ) -> KubectlOutput:
 
             try:
+                # 设置CS客户端
+                self._setup_cs_client(ctx)
+
                 # 检查是否为只读模式
                 if not self.allow_write:
                     is_write_command, not_allow_write_error = self.is_write_command(command)
@@ -304,8 +536,9 @@ async def ack_kubectl(
                         exit_code=1
                     )
 
-                # 获取本机 kubeconfig 文件路径
-                kubeconfig_path = self._get_kubeconfig_path()
+                # 获取 kubeconfig 文件路径
+                context_manager = get_context_manager()
+                kubeconfig_path = context_manager.get_kubeconfig_path(cluster_id)
 
                 # 检查是否为流式命令
                 is_streaming, stream_type = self.is_streaming_command(command)
