Build an AI Kubernetes Troubleshooting Agent

Build an AI Kubernetes Troubleshooting Agent

Debugging a broken pod is a loop: check status, read events, pull logs, form a hypothesis, check the next thing. The loop is mechanical. The judgment in the middle is not, and that middle is exactly what an LLM is good at.

This post builds an agent that runs that loop. You describe the symptom, it investigates the cluster on its own and reports what's wrong. It reads pod status, events, and logs — and it cannot write anything, because we make that structurally impossible rather than asking it nicely.

That last point is where most "AI agent for Kubernetes" demos go wrong, so it's where we'll spend the most time.

What we're building

$ python agent.py "checkout-api in prod keeps restarting"

[tool] get_pods(namespace=prod, selector=app=checkout-api)
[tool] get_events(namespace=prod, name=checkout-api-7d9f8-x2kql)
[tool] get_logs(namespace=prod, name=checkout-api-7d9f8-x2kql, previous=True)

The container is being OOMKilled. Its memory limit is 256Mi, but the previous
container's logs end mid-request during a bulk export, and the last event is
"Memory cgroup out of memory". Exit code 137 confirms the kernel killed it.

Raise the limit to at least 512Mi, or page the export instead of loading it
into memory. The restart count is 47 over 2 hours, so this is not a cold-start
problem.

Around 200 lines of Python. The agent decides which tools to call and in what order — we never script the investigation.

The shape of an agent

An "agent" here is one loop:

1. Send the conversation plus a list of tools to the model.
2. If the response has stop_reason == "tool_use", run the requested tools.
3. Append the results and go back to step 1.
4. When stop_reason == "end_turn", the model is done. Print the answer.

That's it. Everything else — the framework, the orchestration layer, the "agentic" branding — is decoration on this loop. Writing it by hand once is worth more than reading three framework tutorials, because when an agent misbehaves in production you'll be debugging this loop.

Prerequisites

pip install anthropic kubernetes
export ANTHROPIC_API_KEY=sk-ant-...

Plus a kubeconfig pointing at a cluster. Use a dev cluster for your first run.

Step 1: The tools

Tools are ordinary Python functions plus a JSON Schema telling the model when to call them. Our agent gets four, and every one is read-only.

1from datetime import datetime, timezone
2
3from kubernetes import client, config
4
5config.load_kube_config()
6core = client.CoreV1Api()
7
8MAX_LOG_BYTES = 20_000
9UNDATED = datetime.min.replace(tzinfo=timezone.utc)
10
11
12def get_pods(namespace: str, selector: str | None = None) -> str:
13    pods = core.list_namespaced_pod(namespace, label_selector=selector or "")
14    if not pods.items:
15        return f"No pods found in {namespace}" + (f" matching {selector}" if selector else "")
16
17    lines = []
18    for p in pods.items:
19        statuses = p.status.container_statuses or []
20        restarts = sum(c.restart_count for c in statuses)
21        waiting = [
22            c.state.waiting.reason
23            for c in statuses
24            if c.state and c.state.waiting and c.state.waiting.reason
25        ]
26        terminated = [
27            f"exit {c.last_state.terminated.exit_code} ({c.last_state.terminated.reason})"
28            for c in statuses
29            if c.last_state and c.last_state.terminated
30        ]
31        detail = ", ".join(waiting + terminated)
32        lines.append(
33            f"{p.metadata.name}  phase={p.status.phase}  restarts={restarts}"
34            + (f"  {detail}" if detail else "")
35        )
36    return "\n".join(lines)
37
38
39def get_events(namespace: str, name: str | None = None) -> str:
40    field = f"involvedObject.name={name}" if name else ""
41    events = core.list_namespaced_event(namespace, field_selector=field)
42    # Some events carry neither timestamp; a bare 0 would compare int
43    # against datetime and raise TypeError.
44    rows = sorted(events.items,
45                  key=lambda e: e.last_timestamp or e.event_time or UNDATED)
46    if not rows:
47        return "No events found."
48    # Events are chronological; the last ones are the ones that matter.
49    return "\n".join(
50        f"{e.last_timestamp}  {e.type}  {e.reason}: {e.message}" for e in rows[-25:]
51    )
52
53
54def get_logs(namespace: str, name: str, container: str | None = None,
55             previous: bool = False) -> str:
56    try:
57        text = core.read_namespaced_pod_log(
58            name=name, namespace=namespace, container=container,
59            previous=previous, tail_lines=200,
60            limit_bytes=MAX_LOG_BYTES,   # capped server-side, not after transfer
61        )
62    except client.exceptions.ApiException as e:
63        return f"Could not read logs: {e.reason}"
64
65    return text or "(empty)"
66
67
68def describe_pod(namespace: str, name: str) -> str:
69    p = core.read_namespaced_pod(name=name, namespace=namespace)
70    spec = p.spec.containers[0]
71    return (
72        f"image: {spec.image}\n"
73        f"resources: {spec.resources.to_dict() if spec.resources else 'none'}\n"
74        f"node: {p.spec.node_name}\n"
75        f"conditions: " + ", ".join(
76            f"{c.type}={c.status}" for c in (p.status.conditions or [])
77        )
78    )

Three details in there are load-bearing.

previous=True on logs. For a crashing container, the current log is the one that just started and hasn't failed yet. The evidence is in the previous container's logs. This is the single most common mistake when debugging restart loops by hand too — see Fix Kubernetes CrashLoopBackOff for the manual version of the same investigation.

tail_lines and limit_bytes together. A chatty pod produces megabytes of logs. Every byte a tool returns becomes input tokens on the next API call, and the one after that, for the rest of the conversation. An unbounded log tool is a token bomb: one call can cost more than the entire rest of the investigation.

Both caps are parameters on the API call, so the truncation happens on the server and the bytes never cross the network. Truncating in Python after the fact works too, but you've already paid to transfer the 40 MB.

get_pods returns a digest, not the object. A full PodList as JSON is thousands of tokens of metadata.managedFields the model will never use. Give it phase, restart count, and the waiting/terminated reason — the four things that actually drive the diagnosis.

Declaring the tools

1TOOLS = [
2    {
3        "name": "get_pods",
4        "description": (
5            "List pods in a namespace with phase, restart count, and failure reason. "
6            "Start here when you don't yet know which pod is broken."
7        ),
8        "input_schema": {
9            "type": "object",
10            "properties": {
11                "namespace": {"type": "string"},
12                "selector": {
13                    "type": "string",
14                    "description": "Optional label selector, e.g. app=checkout-api",
15                },
16            },
17            "required": ["namespace"],
18        },
19    },
20    {
21        "name": "get_events",
22        "description": (
23            "Recent events for a namespace or a single object. Events explain "
24            "scheduling failures, image pull errors, and OOM kills."
25        ),
26        "input_schema": {
27            "type": "object",
28            "properties": {
29                "namespace": {"type": "string"},
30                "name": {"type": "string", "description": "Object name to filter on"},
31            },
32            "required": ["namespace"],
33        },
34    },
35    {
36        "name": "get_logs",
37        "description": (
38            "Last 200 lines of container logs. Set previous=true to read the logs of "
39            "the container that just crashed — for a restart loop that is where the "
40            "error is."
41        ),
42        "input_schema": {
43            "type": "object",
44            "properties": {
45                "namespace": {"type": "string"},
46                "name": {"type": "string"},
47                "container": {"type": "string"},
48                "previous": {"type": "boolean"},
49            },
50            "required": ["namespace", "name"],
51        },
52    },
53    {
54        "name": "describe_pod",
55        "description": "Image, resource requests/limits, node, and conditions for one pod.",
56        "input_schema": {
57            "type": "object",
58            "properties": {
59                "namespace": {"type": "string"},
60                "name": {"type": "string"},
61            },
62            "required": ["namespace", "name"],
63        },
64    },
65]
66
67DISPATCH = {
68    "get_pods": get_pods,
69    "get_events": get_events,
70    "get_logs": get_logs,
71    "describe_pod": describe_pod,
72}

Write the descriptions for the model, not for a human reader. "Set previous=true … that is where the error is" is not documentation — it's the instruction that stops the agent reading the wrong log. Tool descriptions are the highest-leverage prompt real estate you have, because they're consulted at exactly the moment the decision is made.

Step 2: The loop

1from anthropic import Anthropic
2
3anthropic = Anthropic()
4
5SYSTEM = """You are a Kubernetes troubleshooting agent with read-only cluster access.
6
7Investigate the reported symptom by calling tools. Work from the general to the
8specific: find the failing pod, read its events, then read the logs of the
9container that failed.
10
11Report the root cause and the specific fix. Cite the evidence you used — the exit
12code, the event, the log line. If the evidence is not conclusive, say what you
13would need to look at next rather than guessing.
14
15You cannot modify the cluster. Do not suggest that you have applied a fix."""
16
17
18def run(symptom: str, max_turns: int = 12) -> str:
19    messages = [{"role": "user", "content": symptom}]
20
21    for _ in range(max_turns):
22        response = anthropic.messages.create(
23            model="claude-opus-5",
24            max_tokens=16000,
25            system=SYSTEM,
26            tools=TOOLS,
27            messages=messages,
28        )
29
30        if response.stop_reason == "refusal":
31            return "The request was declined by safety classifiers."
32
33        messages.append({"role": "assistant", "content": response.content})
34
35        if response.stop_reason != "tool_use":
36            return "".join(b.text for b in response.content if b.type == "text")
37
38        results = []
39        for block in response.content:
40            if block.type != "tool_use":
41                continue
42            print(f"[tool] {block.name}({format_args(block.input)})")
43            try:
44                output = DISPATCH[block.name](**block.input)
45                results.append({
46                    "type": "tool_result",
47                    "tool_use_id": block.id,
48                    "content": output,
49                })
50            except Exception as e:
51                results.append({
52                    "type": "tool_result",
53                    "tool_use_id": block.id,
54                    "content": f"{type(e).__name__}: {e}",
55                    "is_error": True,
56                })
57
58        messages.append({"role": "user", "content": results})
59
60    return "Hit the turn limit without reaching a conclusion."
61
62
63def format_args(d: dict) -> str:
64    return ", ".join(f"{k}={v}" for k, v in d.items())

Four things here are easy to get wrong.

Append response.content, not the text. The assistant turn has to carry the tool_use blocks back, or the API can't match your results to the calls. Extracting .text and appending that string is the most common bug in a hand-written loop.

Return every result in one user message. The model can request several tools in one turn and they run concurrently. Splitting the results across multiple messages teaches it not to parallelize.

Return tool errors as is_error results — don't raise. A pod that doesn't exist is information the agent can act on ("that name is wrong, let me list the namespace"). Crashing the loop throws that away.

Check stop_reason before reading content. A refusal returns HTTP 200 with an empty or partial content array. Code that indexes content[0] unconditionally raises IndexError on a perfectly successful API call.

max_turns is the circuit breaker. An agent that can't find the answer will keep looking, and every turn resends the whole conversation. Bound it.

One sizing note: on Claude Opus 5 thinking is on by default when you omit the thinking parameter, and max_tokens caps thinking plus response text together. 16,000 is comfortable for this workload, but if you tighten it around the expected answer length you'll truncate mid-response rather than getting a short answer.

Step 3: The RBAC ceiling

Now the part that matters.

The system prompt says "You cannot modify the cluster." That sentence is worth nothing — it's a request, competing with whatever else lands in the context window, including pod logs that any workload can write to. The defense is that the agent's credentials cannot perform a write, so what it decides to do stops mattering.

1apiVersion: v1
2kind: ServiceAccount
3metadata:
4  name: k8s-troubleshooting-agent
5  namespace: agents
6---
7apiVersion: rbac.authorization.k8s.io/v1
8kind: ClusterRole
9metadata:
10  name: troubleshooting-agent-readonly
11rules:
12  - apiGroups: [""]
13    resources: ["pods", "events", "services", "nodes"]
14    verbs: ["get", "list"]
15  - apiGroups: [""]
16    resources: ["pods/log"]
17    verbs: ["get"]
18  - apiGroups: ["apps"]
19    resources: ["deployments", "replicasets", "statefulsets"]
20    verbs: ["get", "list"]
21---
22apiVersion: rbac.authorization.k8s.io/v1
23kind: ClusterRoleBinding
24metadata:
25  name: troubleshooting-agent-readonly
26roleRef:
27  apiGroup: rbac.authorization.k8s.io
28  kind: ClusterRole
29  name: troubleshooting-agent-readonly
30subjects:
31  - kind: ServiceAccount
32    name: k8s-troubleshooting-agent
33    namespace: agents

No create, no update, no patch, no delete, no exec — and no secrets. Verify that rather than trusting it:

SA=system:serviceaccount:agents:k8s-troubleshooting-agent
kubectl auth can-i --list --as=$SA
kubectl auth can-i delete pods --as=$SA   # no
kubectl auth can-i get secrets --as=$SA   # no

Run those in CI. They are the actual security boundary, and unlike a prompt they can be asserted on.

That manifest is the minimum needed to run this agent. Scoping it properly for production — why not to bind to view, why reads are the risk surface once a model provider is downstream, and how to invert your audit policy — is its own subject: Give an AI Agent Read-Only Access to Kubernetes.

The ceiling is also why our tools are four narrow functions instead of one run_kubectl(command) tool. A bash-shaped tool gives the model maximum leverage and gives you an opaque string to authorize. get_logs(namespace, name) is a decision you can inspect, rate-limit, log, and scope to a namespace. kubectl {arbitrary} is not. Give an agent a shell and your only remaining control is the credential it runs as — which is why the RBAC ceiling stops being a defense-in-depth layer and becomes the entire defense.

Step 4: Cost

The loop resends the entire conversation on every turn. Logs and events dominate that payload, so cost grows with the square of the investigation length if you're careless.

Two levers.

Cache the stable prefix. The system prompt and tool definitions are byte-identical on every turn of every run. Mark the end of that prefix and it bills at roughly a tenth of the input rate after the first write:

1response = anthropic.messages.create(
2    model="claude-opus-5",
3    max_tokens=16000,
4    system=[{
5        "type": "text",
6        "text": SYSTEM,
7        "cache_control": {"type": "ephemeral"},
8    }],
9    tools=TOOLS,
10    messages=messages,
11)

Caching is a prefix match — tools render before the system prompt, so a marker on the system block covers both. It also means anything volatile at the front invalidates everything after it. Don't interpolate a timestamp or a cluster name into the system prompt; put dynamic context in the first user message instead. Confirm it's working by checking response.usage.cache_read_input_tokens is non-zero after the first call — if it's stuck at zero, something in your prefix is changing between requests.

Tune effort to the task. Most Kubernetes failures are pattern matches, not research problems. Reserve deep reasoning for the ones that aren't:

output_config={"effort": "low"},   # ImagePullBackOff, obvious OOM
output_config={"effort": "high"},  # intermittent, multi-component failures

An easy escalation policy: run at low, and if the agent reports it can't reach a conclusion, re-run the same symptom at high.

What it's good at, and what it isn't

Two weeks of running this against real clusters, honestly:

It's reliably good at single-pod failures with local evidence — CrashLoopBackOff, OOMKilled, ImagePullBackOff, failing probes, pods stuck Pending on unschedulable resource requests. These are exactly the cases where the answer is in the events and the previous container's logs, and the work is knowing where to look.

It's unreliable at anything needing cluster-wide correlation — a CNI problem showing up as random timeouts across ten services, or a failure whose cause is three hops away in a dependency. It'll produce a confident, plausible, wrong answer. The tools we gave it can't see the shape of that problem, and the model doesn't reliably say "I can't see enough."

It is not an incident responder. It's a first-pass triage that turns "something is broken" into "here's the evidence and the likely cause" in about twenty seconds. That's genuinely useful at 3am, and it is a much smaller claim than most demos in this space make.

The honest framing: this replaces the ten minutes of mechanical kubectl archaeology at the start of an investigation. It does not replace the engineer. For the manual version of that same archaeology — worth knowing, because you'll need it when the agent is wrong — see the Kubernetes debugging and troubleshooting guide.

Where this goes next

Three directions, roughly in order of payoff:

Expose the tools over MCP instead of importing them into one script. The Model Context Protocol turns this tool set into a server any MCP-aware client can use — your editor, a chat client, a larger agent — without re-implementing the Kubernetes plumbing each time. The read-only ServiceAccount stays the boundary. We've written about what MCP changes for DevOps if the protocol is new to you.

Add Prometheus as a fifth tool. Most of the "unreliable" cases above are unreliable because the agent can see one pod's logs but not the request rate, error rate, or saturation around it. A query_prometheus(promql) tool — still read-only — closes a lot of that gap.

Add a verification pass. Have a second call check the first one's conclusion against the collected evidence, with a specific instruction to flag claims the tool output doesn't support. Confident-and-wrong is this agent's main failure mode, and a skeptical reader catches a good share of it.

The through-line in all three: capability comes from better tools and better evidence, not from a longer prompt. The prompt is the smallest part of this system. The RBAC ceiling is the most important one.