Master MCP Servers for DevOps from Zero to Hero: Build an AI Kubernetes Incident Debugger in Python

Stop running ten kubectl commands to debug one failing pod. Build an MCP server that lets an AI do it for you safely, with RBAC

It’s 2:07 AM.

Your phone explodes with alerts.

Production is down. The payment service is crash-looping. Revenue is literally bleeding every minute.

You open your terminal and start the familiar Kubernetes ritual:

kubectl get pods -n production
kubectl describe pod payment-service-7d9f8b-xk2p9 -n production
kubectl logs payment-service-7d9f8b-xk2p9 -n production --previous
kubectl get events -n production --field-selector reason=BackOff
kubectl top pods -n production

One command becomes five.

Five becomes fifteen.

You copy errors into Google. Open three tabs. Check Slack. Run more commands.

Meanwhile, the clock keeps moving.

20 minutes later, you finally find the root cause.

Now imagine this instead:

You ask your AI assistant:

“Why is payment-service crash-looping in production?”

And within 30 seconds, it responds with:

• the failing container
• recent crash logs
• memory spikes
• Kubernetes events
• likely root cause
• recommended fixes

No command hunting. No context switching. No panic-driven debugging.

That’s the power of Model Context Protocol (MCP).

And in this article, you’ll build exactly that:

A production-grade Kubernetes incident debugger powered by MCP and Python

What Is MCP? (In 2 Minutes)

Think of MCP as the USB-C for AI.

A universal standard that lets AI assistants plug into external systems like:

• Kubernetes
• PostgreSQL
• CI/CD pipelines
• internal APIs
• observability platforms
• cloud infrastructure

Without MCP, every AI integration was basically custom plumbing.

Want an AI assistant to inspect Kubernetes logs? Build a connector.

Want GPT-4 to query PostgreSQL? Build another connector.

Want the same integration to work inside Claude, Cursor, VS Code, and your internal agent framework?

You guessed it.

Build it all over again.

The ecosystem was becoming fragmented fast.

Every model vendor had its own tool-calling layer. Every company was reinventing the same integration stack.

So in November 2024, Anthropic open-sourced the Model Context Protocol (MCP).

The goal was simple:

Standardize how AI systems connect to tools, data, and external context.

Instead of building separate integrations for every AI platform, you build one MCP server.

That server exposes:

• tools
• resources
• prompts
• structured context

And any MCP-compatible client can use it.

That includes:

• OpenAI clients
• Claude Desktop
• Cursor
• Visual Studio Code
• LangChain agents
• internal copilots
• autonomous workflows

One protocol. Many clients.

Why That Matters

MCP changes AI from:

“a chatbot that knows text”

into:

“an intelligent system that can actually interact with infrastructure.”

That’s a massive shift.

Because once AI can reliably access:

• logs
• metrics
• databases
• deployments
• APIs
• production systems

…it stops being just an assistant.

It becomes operational.

And MCP Is Bigger Than One Company Now

In December 2025, Anthropic donated MCP to the Linux Foundation’s Agentic AI Foundation (AAIF).

The initiative was co-founded by:

• Anthropic
• Block
• OpenAI

And backed by major players including:

• Google
• Microsoft
• AWS
• Cloudflare
• Bloomberg

That matters for one reason:

MCP is no longer a vendor feature.

It’s becoming an open standard for the next generation of AI systems.

The Three Primitives

An MCP server exposes exactly three types of capabilities:

The AI decides which tools to call, in what order, based on your question. You define what’s available. The protocol handles the rest.

What We’re Building

A Kubernetes Incident Debugger MCP Server with five tools:

k8s-debugger/
├── server.py          # Main MCP server
├── k8s_client.py      # Kubernetes API wrapper
├── pyproject.toml     # Dependencies
└── k8s/
    ├── serviceaccount.yaml
    ├── clusterrole.yaml
    └── deployment.yaml    # For in-cluster deployment

When complete, you’ll be able to ask your AI:

• “What pods are failing in the production namespace?”
• “Show me the last 50 lines of logs from the api-gateway pod”
• “What events happened in the last 10 minutes that could explain this OOMKilled?”

Setup

mkdir k8s-debugger && cd k8s-debugger

# Initialize project (using uv — the modern Python package manager)
uv init
uv add fastmcp kubernetes

Why uv? It’s 10–100x faster than pip and is now the standard for MCP server development. The FastMCP docs recommend it.

Verify your Python and kubeconfig are ready:

python --version        # 3.14+
kubectl cluster-info    # Should show your cluster

Your pyproject.toml should look like this:

[project]
name = "k8s-debugger"
version = "0.1.0"
requires-python = ">=3.14"
dependencies = [
    "fastmcp>=3.2.4",
    "kubernetes>=35.0.0",
]

Part 1: The Kubernetes Client Wrapper

First, a clean wrapper around the kubernetes Python client. This keeps our MCP server readable and the k8s API calls testable.

# k8s_client.py
from kubernetes import client, config
from kubernetes.client.rest import ApiException
from typing import Optional

def load_k8s_config():
    """Load kubeconfig — works both locally and in-cluster."""
    try:
        # In-cluster: uses the pod's ServiceAccount token automatically
        config.load_incluster_config()
    except config.ConfigException:
        # Local: reads ~/.kube/config
        config.load_kube_config()

def get_failing_pods(namespace: str = "default") -> list[dict]:
    """Return all pods NOT in Running or Succeeded state."""
    load_k8s_config()
    v1 = client.CoreV1Api()

    pods = v1.list_namespaced_pod(namespace=namespace)
    failing = []

    for pod in pods.items:
        phase = pod.status.phase
        if phase not in ("Running", "Succeeded"):
            restart_count = sum(
                cs.restart_count
                for cs in (pod.status.container_statuses or [])
            )
            failing.append({
                "name": pod.metadata.name,
                "namespace": pod.metadata.namespace,
                "phase": phase,
                "restart_count": restart_count,
                "node": pod.spec.node_name,
            })

        # Also catch crash-looping Running pods
        elif phase == "Running":
            for cs in (pod.status.container_statuses or []):
                if cs.restart_count > 5:
                    failing.append({
                        "name": pod.metadata.name,
                        "namespace": pod.metadata.namespace,
                        "phase": f"Running (crash-looping, {cs.restart_count} restarts)",
                        "restart_count": cs.restart_count,
                        "node": pod.spec.node_name,
                    })
                    break

    return failing

def get_pod_logs(
    pod_name: str,
    namespace: str = "default",
    container: Optional[str] = None,
    previous: bool = False,
    tail_lines: int = 100,
) -> str:
    """Fetch pod logs. Set previous=True to get logs from the last crashed container."""
    load_k8s_config()
    v1 = client.CoreV1Api()

    try:
        logs = v1.read_namespaced_pod_log(
            name=pod_name,
            namespace=namespace,
            container=container,
            previous=previous,
            tail_lines=tail_lines,
        )
        return logs or "(no logs available)"
    except ApiException as e:
        if e.status == 404:
            return f"Pod '{pod_name}' not found in namespace '{namespace}'"
        return f"Error fetching logs: {e.reason}"

def describe_pod(pod_name: str, namespace: str = "default") -> dict:
    """Return structured pod description — conditions, container states, resource limits."""
    load_k8s_config()
    v1 = client.CoreV1Api()

    try:
        pod = v1.read_namespaced_pod(name=pod_name, namespace=namespace)
    except ApiException as e:
        return {"error": f"Pod not found: {e.reason}"}

    containers = []
    for cs in (pod.status.container_statuses or []):
        state = cs.state
        state_str = "unknown"
        reason = None

        if state.running:
            state_str = "running"
        elif state.waiting:
            state_str = "waiting"
            reason = state.waiting.reason
        elif state.terminated:
            state_str = "terminated"
            reason = state.terminated.reason

        containers.append({
            "name": cs.name,
            "ready": cs.ready,
            "restart_count": cs.restart_count,
            "state": state_str,
            "reason": reason,
        })

    conditions = [
        {"type": c.type, "status": c.status, "reason": c.reason}
        for c in (pod.status.conditions or [])
    ]

    return {
        "name": pod.metadata.name,
        "namespace": pod.metadata.namespace,
        "phase": pod.status.phase,
        "node": pod.spec.node_name,
        "conditions": conditions,
        "containers": containers,
        "labels": pod.metadata.labels,
    }

def get_namespace_events(
    namespace: str = "default",
    reason_filter: Optional[str] = None,
    limit: int = 20,
) -> list[dict]:
    """Fetch recent warning events. Optionally filter by reason (e.g. 'OOMKilling', 'BackOff')."""
    load_k8s_config()
    v1 = client.CoreV1Api()

    events = v1.list_namespaced_event(
        namespace=namespace,
        field_selector="type=Warning",
    )

    results = []
    for e in events.items:
        if reason_filter and reason_filter.lower() not in e.reason.lower():
            continue
        results.append({
            "reason": e.reason,
            "message": e.message,
            "object": f"{e.involved_object.kind}/{e.involved_object.name}",
            "count": e.count,
            "last_seen": str(e.last_timestamp),
        })

    # Most recent first
    results.sort(key=lambda x: x["last_seen"], reverse=True)
    return results[:limit]

def get_resource_usage(namespace: str = "default") -> list[dict]:
    """Fetch CPU/memory usage for pods via the Metrics API (requires metrics-server)."""
    load_k8s_config()
    api = client.CustomObjectsApi()

    try:
        metrics = api.list_namespaced_custom_object(
            group="metrics.k8s.io",
            version="v1beta1",
            namespace=namespace,
            plural="pods",
        )
    except ApiException as e:
        if e.status == 503:
            return [{"error": "metrics-server is not available. On EKS, install it with: kubectl apply -f https://github.com/kubernetes-sigs/metrics-server/releases/latest/download/components.yaml"}]
        if e.status == 404:
            return [{"error": "metrics.k8s.io APIService not found — metrics-server is not installed in this cluster."}]
        return [{"error": f"Metrics API error: {e.status} {e.reason}"}]
    except Exception as e:
        return [{"error": f"Cannot reach cluster: {e}"}]

    results = []
    for item in metrics.get("items", []):
        pod_name = item["metadata"]["name"]
        for container in item["containers"]:
            results.append({
                "pod": pod_name,
                "container": container["name"],
                "cpu": container["usage"]["cpu"],
                "memory": container["usage"]["memory"],
            })
    return results

def list_namespaces() -> list[str]:
    """Return all namespace names in the cluster."""
    load_k8s_config()
    v1 = client.CoreV1Api()
    return [ns.metadata.name for ns in v1.list_namespace().items]

Part 2: The MCP Server

Now the MCP server itself. This is where the magic happens the @mcp.tool() decorator is all you need to expose a function to any AI client.

# server.py
from fastmcp import FastMCP
from k8s_client import (
    get_failing_pods,
    get_pod_logs,
    describe_pod,
    get_namespace_events,
    get_resource_usage,
    list_namespaces,
)
from typing import Optional
import json

# ── Server initialization ─────────────────────────────────────────────────────
mcp = FastMCP(
    name="k8s-incident-debugger",
    instructions="""
    You are a Kubernetes incident debugger. When asked about cluster issues:
    1. First call list_failing_pods to get an overview of the problem
    2. Use describe_pod to understand the state of specific failing pods
    3. Use get_pod_logs (with previous=true) to read crash logs
    4. Use get_events to surface related Warning events
    5. Correlate the evidence and provide a structured root cause diagnosis
    Always suggest a remediation step, not just a diagnosis.
    """,
)

# ── Tools ─────────────────────────────────────────────────────────────────────

@mcp.tool()
def list_failing_pods(namespace: str = "default") -> str:
    """
    List all pods that are NOT in Running or Succeeded state,
    plus any Running pods with excessive restart counts (crash-looping).

    Args:
        namespace: Kubernetes namespace to check. Use 'all' to check all namespaces.
    """
    if namespace == "all":
        namespaces = list_namespaces()
        all_failing = []
        for ns in namespaces:
            all_failing.extend(get_failing_pods(ns))
        failing = all_failing
    else:
        failing = get_failing_pods(namespace)

    if not failing:
        return f"✅ No failing pods found in namespace '{namespace}'"

    return json.dumps(failing, indent=2, default=str)

@mcp.tool()
def get_logs(
    pod_name: str,
    namespace: str = "default",
    container: Optional[str] = None,
    previous: bool = False,
    tail_lines: int = 100,
) -> str:
    """
    Fetch logs from a specific pod.

    Args:
        pod_name: Name of the pod
        namespace: Kubernetes namespace
        container: Container name (required if pod has multiple containers)
        previous: Set True to get logs from the previously crashed container instance
        tail_lines: Number of lines to return from the end of the log (default 100)
    """
    return get_pod_logs(pod_name, namespace, container, previous, tail_lines)

@mcp.tool()
def inspect_pod(pod_name: str, namespace: str = "default") -> str:
    """
    Get a structured description of a pod: phase, container states,
    restart counts, readiness conditions, and node placement.
    Use this before fetching logs to understand what type of failure occurred.

    Args:
        pod_name: Name of the pod to inspect
        namespace: Kubernetes namespace
    """
    result = describe_pod(pod_name, namespace)
    return json.dumps(result, indent=2, default=str)

@mcp.tool()
def get_events(
    namespace: str = "default",
    reason_filter: Optional[str] = None,
    limit: int = 20,
) -> str:
    """
    Fetch recent Warning events from a namespace.
    Optionally filter by event reason to find specific issues.

    Common reason values: OOMKilling, BackOff, Failed, Unhealthy,
    FailedScheduling, FailedMount, Evicted

    Args:
        namespace: Kubernetes namespace
        reason_filter: Optional string to filter events by reason (case-insensitive)
        limit: Maximum number of events to return (default 20)
    """
    events = get_namespace_events(namespace, reason_filter, limit)
    if not events:
        return f"No warning events found in namespace '{namespace}'"
    return json.dumps(events, indent=2, default=str)

@mcp.tool()
def get_pod_resource_usage(namespace: str = "default") -> str:
    """
    Fetch current CPU and memory usage for all pods in a namespace.
    Requires metrics-server to be installed in the cluster.
    Use this to identify resource-hungry pods or potential OOMKill candidates.

    Args:
        namespace: Kubernetes namespace to check
    """
    usage = get_resource_usage(namespace)
    if not usage:
        return f"No metrics available for namespace '{namespace}'"
    return json.dumps(usage, indent=2, default=str)

# ── Resources ─────────────────────────────────────────────────────────────────

@mcp.resource("cluster://namespaces")
def cluster_namespaces() -> str:
    """Lists all namespaces available in the cluster."""
    namespaces = list_namespaces()
    return "\n".join(namespaces)

# ── Prompts ───────────────────────────────────────────────────────────────────

@mcp.prompt()
def incident_investigation(namespace: str, symptom: str) -> str:
    """
    Structured incident investigation workflow prompt.
    Guides the AI through a systematic debugging process.

    Args:
        namespace: The namespace where the incident is occurring
        symptom: Brief description of the symptom (e.g. 'pods crash-looping', 'service unreachable')
    """
    return f"""
    You are investigating a Kubernetes incident. Symptom: {symptom}
    Namespace: {namespace}

    Follow this investigation sequence:
    1. Call list_failing_pods(namespace="{namespace}") to identify affected pods
    2. For each failing pod, call inspect_pod to understand the failure mode
    3. For pods with crash loops, call get_logs with previous=true to read the crash logs
    4. Call get_events(namespace="{namespace}") to find related Warning events
    5. Synthesize findings into:
       - Root cause (with evidence)
       - Immediate remediation (what to run right now)
       - Prevention (what to fix long-term)
    """

# ── Entry point ───────────────────────────────────────────────────────────────

if __name__ == "__main__":
    import os
    transport = os.getenv("MCP_TRANSPORT", "stdio")

    if transport == "http":
        mcp.run(transport="http", host="0.0.0.0", port=8080)
    else:
        mcp.run()  # stdio for local Claude Desktop / Cursor use

Part 3: Test It Locally With MCP Inspector

Before connecting to any AI client, verify your server works:

# Run the MCP Inspector — a browser-based tool for testing MCP servers
npx -y @modelcontextprotocol/inspector uv run server.py

Open http://127.0.0.1:6274 in your browser. You'll see:

• Tools tab: All five tools listed with their parameters
• Resources tab: The cluster://namespaces resource
• Prompts tab: The incident_investigation prompt template

Click List Failing Pods, set namespace to default, hit Execute. You should see real output from your cluster.

If you don’t have failing pods to test with: kubectl run crasher --image=busybox --restart=OnFailure -- /bin/sh -c "exit 1" creates one instantly.

Part 4: Connect to Claude Desktop

Add the server to your Claude Desktop configuration/Claude code mcp servers:

{
  "mcpServers": {
    "k8s-debugger": {
      "command": "uv",
      "args": [
        "run",
        "/absolute/path/to/k8s-debugger/server.py"
      ]
    }
  }
}

or you can use cli as well to add mpc server

claude mcp add k8s-debugger -- uv run /absolute/path/to/k8s-debugger/server.py

check using /mcp command after lauching claude code or kiro-cli etc

Now ask Claude:

"Check all namespaces for failing pods and tell me what's wrong"

Watch Claude call list_failing_pods, then inspect_pod, then get_logsautonomously, in sequence — and return a structured diagnosis.

Part 5: Deploy In-Cluster (Production Mode)

Running the MCP server on your laptop means AI access dies when you close your terminal. For production, deploy it inside the cluster.

Step 1: RBAC — Read-Only ServiceAccount

The MCP server only needs to read cluster state. Enforce this with a minimal ClusterRole:

# k8s/serviceaccount.yaml
apiVersion: v1
kind: Namespace
metadata:
  name: platform-tools
---
apiVersion: v1
kind: ServiceAccount
metadata:
  name: k8s-debugger-mcp
  namespace: platform-tools
---
# k8s/clusterrole.yaml
apiVersion: rbac.authorization.k8s.io/v1
kind: ClusterRole
metadata:
  name: k8s-debugger-readonly
rules:
  - apiGroups: [""]
    resources: ["pods", "pods/log", "events", "namespaces", "nodes"]
    verbs: ["get", "list", "watch"]
  - apiGroups: ["apps"]
    resources: ["deployments", "replicasets", "daemonsets", "statefulsets"]
    verbs: ["get", "list"]
  - apiGroups: ["metrics.k8s.io"]
    resources: ["pods", "nodes"]
    verbs: ["get", "list"]
---
apiVersion: rbac.authorization.k8s.io/v1
kind: ClusterRoleBinding
metadata:
  name: k8s-debugger-readonly-binding
roleRef:
  apiGroup: rbac.authorization.k8s.io
  kind: ClusterRole
  name: k8s-debugger-readonly
subjects:
  - kind: ServiceAccount
    name: k8s-debugger-mcp
    namespace: platform-tools

Security note: This ClusterRole intentionally omits secrets from the resource list. You do NOT want your MCP server and by extension, the AI model to have access to Kubernetes secrets.

Step 2: Dockerfile

FROM python:3.12-slim

WORKDIR /app
COPY pyproject.toml uv.lock ./
RUN pip install uv && uv sync --frozen --no-dev

COPY server.py k8s_client.py ./

ENV MCP_TRANSPORT=http
EXPOSE 8080

CMD ["uv", "run", "server.py"]

Step 3: Kubernetes Deployment

# k8s/deployment.yaml
apiVersion: apps/v1
kind: Deployment
metadata:
  name: k8s-debugger-mcp
  namespace: platform-tools
spec:
  replicas: 1
  selector:
    matchLabels:
      app: k8s-debugger-mcp
  template:
    metadata:
      labels:
        app: k8s-debugger-mcp
    spec:
      serviceAccountName: k8s-debugger-mcp
      containers:
        - name: mcp-server
          image: your-registry/k8s-debugger-mcp:latest
          ports:
            - containerPort: 8080
          env:
            - name: MCP_TRANSPORT
              value: "http"
          resources:
            requests:
              cpu: 100m
              memory: 128Mi
            limits:
              cpu: 500m
              memory: 256Mi
          readinessProbe:
            tcpSocket:
              port: 8080
            initialDelaySeconds: 5
---
apiVersion: v1
kind: Service
metadata:
  name: k8s-debugger-mcp
  namespace: platform-tools
spec:
  selector:
    app: k8s-debugger-mcp
  ports:
    - port: 8080
      targetPort: 8080

Deploy it:

kubectl apply -f k8s/

Now any MCP-compatible client in your network can connect to [http://k8s-debugger-mcp.platform-tools.svc.cluster.local:8080](http://k8s-debugger-mcp.platform-tools.svc.cluster.local:8080.)

The Security Mental Model

The pattern you’ve built enforces a critical principle: the AI never touches your cluster directly.

User question
     │
     ▼
AI Client (Claude / Cursor / your agent)
     │  MCP protocol (JSON-RPC)
     ▼
k8s-debugger MCP Server   ← Your code. Your RBAC. Your rules.
     │  kubernetes Python client
     ▼
Kubernetes API Server      ← Read-only ServiceAccount
     │
     ▼
Pods, Events, Logs

The MCP server is the only thing that talks to Kubernetes. It has read-only permissions. It filters what it returns. The AI sees exactly what you decide to show it nothing more.

If you later add a restart_deployment tool, you add it explicitly, with its own RBAC role. There's no accidental escalation.

What You Can Add Next

The five tools here are the foundation. Here’s what production teams build on top:

The Bigger Picture

MCP is not just a developer curiosity. It’s the standard integration layer between AI and infrastructure — the same way REST became the standard for web APIs. Now governed by the Linux Foundation’s Agentic AI Foundation, it has buy-in from every major AI lab and cloud provider.

Every MCP server you build is a reusable capability. Your Kubernetes debugger can be used by Claude Desktop today, by a VS Code extension tomorrow, and by your custom incident response agent next month — without changing a line of server code.

The teams that build these integrations now will have dramatically shorter incident response times than the teams still running commands manually in 2027.

You’ve built the foundation. The rest is tools.