updates to tool

Signed-off-by: vsoch <vsoch@users.noreply.github.com>

Author: vsoch

hpc_mcp/deploy/kubernetes/__init__.py

@@ -11,9 +11,15 @@
     kubectl_explain,
     kubectl_get,
     kubectl_get_autoscaler_info,
+    kubectl_get_current_context,
     kubectl_get_events,
     kubectl_get_priority_classes,
     kubectl_get_resource_quota,
+    kubectl_get_summarized,
     kubectl_label,
     kubectl_logs,
+    kubectl_patch,
+    kubectl_rollout_status,
+    kubectl_top_nodes,
+    kubectl_wait,
 )

hpc_mcp/deploy/kubernetes/events.py

@@ -3,6 +3,7 @@
 from typing import Any, Awaitable, Callable, Dict
 
 from kubernetes_asyncio import client, config, watch
+from kubernetes_asyncio.client import ApiClient
 
 
 class KubernetesEvents:
@@ -24,7 +25,7 @@ def get_metadata(self) -> Dict[str, Any]:
         The Agent uses this to know what parameters to send.
         """
         return {
-            "name": "kubernetes_events",
+            "name": "KubernetesEvents",
             "description": "Subscribe to events for Kubernetes resources (Pods, Deployments, etc.)",
             "parameters": {
                 "resource_type": "string (e.g., pods, deployments, kustomizations)",
@@ -43,9 +44,8 @@ async def subscribe(
         sub_id = f"k8s_{uuid.uuid4().hex[:8]}"
 
         # Spin up the background listener
-        task = asyncio.create_task(self._watch_loop(sub_id, params, callback))
+        task = asyncio.create_task(self.watch_loop(sub_id, params, callback))
         self._active_watches[sub_id] = task
-
         return sub_id
 
     async def unsubscribe(self, sub_id: str) -> bool:
@@ -62,46 +62,97 @@ async def unsubscribe(self, sub_id: str) -> bool:
             return True
         return False
 
-    async def _watch_loop(self, sub_id: str, params: Dict[str, Any], callback: Callable):
+    async def watch_loop(self, sub_id: str, params: Dict[str, Any], callback: Callable):
         """
         The internal async loop that communicates with the K8s API.
         """
-        resource_type = params.get("resource_type", "pods")
+        resource_type = params.get("resource_type", "pod").lower()
         namespace = params.get("namespace", "default")
-
-        # Dynamic API selection based on resource type
-        v1 = client.CoreV1Api()
+        is_custom = params.get("is_custom_resource", False)
+        label_selector = params.get("label_selector", "")
+        field_selector = params.get("field_selector", "")
         w = watch.Watch()
 
         try:
-            # Note: This is an example for Pods.
-            # You would extend this to support Deployments/Flux CRDs.
-            method = getattr(v1, f"list_namespaced_{resource_type}")
-
-            async with w.stream(
-                method,
-                namespace=namespace,
-                label_selector=params.get("label_selector", ""),
-                field_selector=params.get("field_selector", ""),
-            ) as stream:
+            if is_custom:
+                api_client_inst = client.CustomObjectsApi()
+                group = params.get("group")
+                version = params.get("version")
+                plural = params.get("plural", resource_type + "s")
+                if not group or not version:
+                    raise ValueError("Both 'group' and 'version' are required.")
+
+                method = api_client_inst.list_namespaced_custom_object
+                stream_args = {
+                    "group": group,
+                    "version": version,
+                    "namespace": namespace,
+                    "plural": plural,
+                }
+            else:
+                if resource_type == "job":
+                    api_client_inst = client.BatchV1Api()
+                else:
+                    api_client_inst = client.CoreV1Api()
+
+                method = getattr(api_client_inst, f"list_namespaced_{resource_type}")
+                stream_args = {
+                    "namespace": namespace,
+                }
+
+            stream_args["label_selector"] = label_selector
+            stream_args["field_selector"] = field_selector
+
+            async with w.stream(method, **stream_args) as stream:
+                sanitizer = ApiClient()
                 async for event in stream:
-                    # Clean up the object for the LLM to save tokens
-                    obj = event["raw_object"]
+                    obj = sanitizer.sanitize_for_serialization(event["object"])
                     event_data = {
                         "type": event["type"],
-                        "name": obj["metadata"]["name"],
-                        "status": obj.get("status", {}),
+                        "name": obj["metadata"].get("name"),
                         "resource": resource_type,
+                        "creation_timestamp": obj["metadata"].get("creationTimestamp"),
                     }
 
-                    # Push notification to MCP client via the provided callback
-                    await callback(sub_id, event_data)
+                    status_block = obj.get("status", {})
+
+                    if resource_type == "pod":
+                        event_data["status"] = status_block.get("phase", "Unknown")
+
+                    elif resource_type == "job":
+                        event_data["status"] = {
+                            "active": int(status_block.get("active", 0)),
+                            "succeeded": int(status_block.get("succeeded", 0)),
+                            "failed": int(status_block.get("failed", 0)),
+                            "start_time": status_block.get("startTime"),
+                        }
+                        completions = obj.get("spec", {}).get("completions", 1)
+                        if event_data["status"]["succeeded"] >= completions:
+                            event_data["state"] = "Succeeded"
+                        elif event_data["status"]["failed"] > 0:
+                            event_data["state"] = "Failed"
+                        else:
+                            event_data["state"] = "Running"
+                    else:
+                        event_data["status"] = status_block
+
+                    if is_custom and "state" in status_block:
+                        event_data["state"] = status_block["state"]
+
+                    try:
+                        safe_event_data = json.loads(json.dumps(event_data, default=str))
+                    except:
+                        safe_event_data = event_data
+
+                    print(f"📡 {sub_id}: {safe_event_data['name']} -> {event['type']}")
+                    await callback(sub_id, safe_event_data)
 
         except asyncio.CancelledError:
-            # Clean exit on unsubscribe
-            pass
+            print(f"🛑 Stopping watcher for {sub_id}...")
         except Exception as e:
-            # Notify the agent that the watch failed
-            await callback(sub_id, {"error": str(e), "status": "failed"})
+            print(f"❌ Watch failed for subscription {sub_id}: {e}")
+            import traceback
+
+            traceback.print_exc()
         finally:
             w.stop()