Add DevOps tools to Gemini orchestrator

k8s/agents-api.yaml

@@ -1,3 +1,34 @@
+apiVersion: v1
+kind: ServiceAccount
+metadata:
+  name: agents-ops-sa
+  namespace: agents-runtime
+---
+apiVersion: rbac.authorization.k8s.io/v1
+kind: ClusterRole
+metadata:
+  name: agents-ops-clusterrole
+rules:
+- apiGroups: ["", "apps", "batch", "networking.k8s.io"]
+  resources: ["pods", "pods/log", "deployments", "services", "namespaces", "events", "jobs", "cronjobs", "ingresses"]
+  verbs: ["get", "list", "watch"]
+- apiGroups: [""]
+  resources: ["pods"]
+  verbs: ["delete"]
+---
+apiVersion: rbac.authorization.k8s.io/v1
+kind: ClusterRoleBinding
+metadata:
+  name: agents-ops-clusterrolebinding
+roleRef:
+  apiGroup: rbac.authorization.k8s.io
+  kind: ClusterRole
+  name: agents-ops-clusterrole
+subjects:
+- kind: ServiceAccount
+  name: agents-ops-sa
+  namespace: agents-runtime
+---
 apiVersion: apps/v1
 kind: Deployment
 metadata:
@@ -14,6 +45,7 @@ spec:
       labels:
         app.kubernetes.io/name: agents-api
     spec:
+      serviceAccountName: agents-ops-sa
       initContainers:
         - name: git-clone
           image: alpine/git

requirements.txt

@@ -6,6 +6,7 @@ langchain-anthropic
 temporalio
 pydantic
 langgraph
+kubernetes
 
 SQLAlchemy==2.0.29
 psycopg2-binary==2.9.9

src/agent.py

@@ -55,7 +55,34 @@ def trigger_utility_scraper() -> str:
         print(f"Failed to trigger utility scraper: {e}")
         return f"Failed to trigger utility scraper: {str(e)}"
 
-tools = [send_whatsapp, set_reminder, schedule_calendar, trigger_utility_scraper]
+from kubernetes import client, config
+
+@tool
+def get_k8s_pods(namespace: str) -> str:
+    """Get the status of all pods in a specific Kubernetes namespace."""
+    try:
+        config.load_incluster_config()
+        v1 = client.CoreV1Api()
+        pods = v1.list_namespaced_pod(namespace)
+        result = f"Pods in {namespace}:\n"
+        for p in pods.items:
+            result += f"- {p.metadata.name} (Status: {p.status.phase})\n"
+        return result
+    except Exception as e:
+        return f"K8s Error: {str(e)}"
+
+@tool
+def delete_k8s_pod(pod_name: str, namespace: str) -> str:
+    """Delete a specific pod in a Kubernetes namespace to restart it."""
+    try:
+        config.load_incluster_config()
+        v1 = client.CoreV1Api()
+        v1.delete_namespaced_pod(name=pod_name, namespace=namespace)
+        return f"Successfully deleted pod {pod_name} in {namespace}. It will automatically restart."
+    except Exception as e:
+        return f"K8s Error: {str(e)}"
+
+tools = [send_whatsapp, set_reminder, schedule_calendar, trigger_utility_scraper, get_k8s_pods, delete_k8s_pod]
 
 @activity.defn
 def run_agent_activity(payload: dict) -> str:
@@ -72,6 +99,8 @@ Their WhatsApp Group/Chat ID (target_jid) is: {target_jid}.
 Whenever you need to send a message back, use the `send_whatsapp` tool and pass {target_jid} as the target_jid!
 If the user asks to scrape utility data, run the utility scraper, or fetch utility bills/data, call the `trigger_utility_scraper` tool.
 If they ask for a reminder, use the `set_reminder` tool and pass {target_jid} as the target_jid!
+If they ask to check Kubernetes pods, use the `get_k8s_pods` tool.
+If they ask to restart a pod or deployment, use the `delete_k8s_pod` tool.
 """
     
     # Use state_modifier or prompt depending on langgraph version, falling back to older syntax if needed.