We are building a command-line agent that troubleshoots failing Kubernetes deployments by calling diagnostic tools and reasoning over the results. If you run services in production and are tired of copy-pasting kubectl commands at 2 a.m., this automates the first 10 minutes of incident response.
What you'll need
- Python 3.10 or later
- An Oxlo.ai API key from https://portal.oxlo.ai
- The OpenAI SDK:
pip install openai
Step 1: Mock the cluster environment
I do not want to require a live cluster, so I will stub three common kubectl operations. These functions return realistic failure scenarios that our agent will investigate.
# k8s_mock.py
import json
def check_pod_status(namespace="default"):
return json.dumps({
"namespace": namespace,
"pods": [
{
"name": "api-gateway-7c9b8f4c5-x2vlp",
"status": "CrashLoopBackOff",
"restarts": 42,
"age": "3h"
},
{
"name": "api-gateway-7c9b8f4c5-abc12",
"status": "Running",
"restarts": 0,
"age": "3h"
}
]
})
def get_recent_logs(pod_name, namespace="default"):
if "api-gateway" in pod_name:
return (
"Connection refused to postgres://db.internal:5432/orders\n"
"Traceback (most recent call last):\n"
" File '/app/main.py', line 44, in connect_db\n"
" raise DatabaseConnectionError"
)
return "No logs available."
def describe_deployment(name, namespace="default"):
return json.dumps({
"name": name,
"namespace": namespace,
"replicas": 3,
"available_replicas": 1,
"strategy": "RollingUpdate",
"last_image": "api-gateway:v2.1.4"
})
Step 2: Define the tool schema
Oxlo.ai models support OpenAI-compatible function calling. We describe the signatures so the model knows exactly what data each tool expects.
# tools.py
TOOLS = [
{
"type": "function",
"function": {
"name": "check_pod_status",
"description": "List pods and their restart counts in a namespace.",
"parameters": {
"type": "object",
"properties": {
"namespace": {
"type": "string",
"description": "Kubernetes namespace to query."
}
},
"required": ["namespace"]
}
}
},
{
"type": "function",
"function": {
"name": "get_recent_logs",
"description": "Fetch the last 50 lines of logs for a specific pod.",
"parameters": {
"type": "object",
"properties": {
"pod_name": {
"type": "string",
"description": "Exact pod name."
},
"namespace": {
"type": "string",
"description": "Kubernetes namespace."
}
},
"required": ["pod_name", "namespace"]
}
}
},
{
"type": "function",
"function": {
"name": "describe_deployment",
"description": "Get deployment metadata including replica counts and image tag.",
"parameters": {
"type": "object",
"properties": {
"name": {
"type": "string",
"description": "Deployment name."
},
"namespace": {
"type": "string",
"description": "Kubernetes namespace."
}
},
"required": ["name", "namespace"]
}
}
}
]
Step 3: Write the system prompt
The prompt keeps the agent focused on evidence. It must not guess the root cause until it has checked pod status, logs, and deployment configuration.
SYSTEM_PROMPT = """You are an SRE troubleshooting agent. Your job is to diagnose why a Kubernetes workload is unhealthy.
Rules:
1. Always start by checking pod status.
2. If a pod is not Running, fetch its logs.
3. Then describe the deployment to check replica skew or image changes.
4. Only after you have tool results, state the likely root cause and give a concrete remediation command or config change.
5. Be concise. Prefer one-line shell commands when possible.
Current date: 2025-01-15
"""
Step 4: Build the agent loop
This loop sends the incident report to Oxlo.ai. I use the Qwen 3 32B model because it handles multi-step agent workflows well, and Oxlo.ai request-based pricing keeps the cost flat even when we append large log payloads back into the conversation. If the response includes tool calls, we execute them and append the results to the history, repeating until the model returns a text answer.
# agent_loop.py
from openai import OpenAI
import json
from k8s_mock import check_pod_status, get_recent_logs, describe_deployment
from tools import TOOLS
client = OpenAI(base_url="https://api.oxlo.ai/v1", api_key="YOUR_OXLO_API_KEY")
TOOL_MAP = {
"check_pod_status": check_pod_status,
"get_recent_logs": get_recent_logs,
"describe_deployment": describe_deployment,
}
def run_agent(user_report: str, namespace: str = "default", deployment: str = "api-gateway"):
messages = [
{"role": "system", "content": SYSTEM_PROMPT},
{
"role": "user",
"content": f"Namespace: {namespace}\nDeployment: {deployment}\nIncident: {user_report}"
},
]
for _ in range(5):
response = client.chat.completions.create(
model="qwen-3-32b",
messages=messages,
tools=TOOLS,
tool_choice="auto",
)
message = response.choices[0].message
if message.content:
print("Agent:", message.content)
if not message.tool_calls:
return message.content
messages.append({
"role": "assistant",
"content": message.content or "",
"tool_calls": [tc.model_dump() for tc in message.tool_calls]
})
for tc in message.tool_calls:
fn_name = tc.function.name
args = json.loads(tc.function.arguments)
print(f"Tool call: {fn_name}({args})")
result = TOOL_MAP[fn_name](**args)
messages.append({
"role": "tool",
"tool_call_id": tc.id,
"content": str(result),
})
return "Reached max iterations without conclusion."
Step 5: Add the entrypoint
I wrap the loop in a small CLI so we can pass in the incident description when we run the script.
# agent.py
import sys
from agent_loop import run_agent
if __name__ == "__main__":
if len(sys.argv) < 2:
print("Usage: python agent.py 'The api-gateway latency is spiking and some pods are restarting.'")
sys.exit(1)
report = sys.argv[1]
run_agent(report)
Run it
Install the dependency and invoke the agent with a realistic incident description.
$ pip install openai
$ export OXLO_API_KEY="sk-oxlo.ai-..."
$ python agent.py "Users are seeing 502 errors and the api-gateway pods keep restarting."
Example output:
Tool call: check_pod_status({'namespace': 'default'})
Tool call: get_recent_logs({'pod_name': 'api-gateway-7c9b8f4c5-x2vlp', 'namespace': 'default'})
Tool call: describe_deployment({'name': 'api-gateway', 'namespace': 'default'})
Agent: Root cause: the api-gateway container cannot connect to the orders database at postgres://db.internal:5432/orders. The pod is crashlooping because of an unhandled DatabaseConnectionError at startup.
Remediation:
1. Verify the DB endpoint is reachable from the cluster.
2. Check if the connection string secret was rotated recently.
3. Roll back to the previous image if v2.1.4 introduced a schema mismatch.
4. Increase the initial delay seconds of the liveness probe so the pod does not restart before the DB retry logic finishes.
Next steps
Swap the mock functions for real subprocess calls to kubectl or the Kubernetes Python client, and wire the script to your paging webhook so it runs automatically when an alert fires. If you want to cut inference costs while the agent reads long log dumps, Oxlo.ai request-based pricing stays flat regardless of how many tokens those logs add to the context window, which makes it a strong fit for agentic loops that grow conversations quickly. Check the details at https://oxlo.ai/pricing.
Top comments (0)