Platform Engineering

Build a Kubernetes MCP Server in Python: A Hands-On Tutorial

Intermediate60 min to complete11 min readAugust 20, 2026Updated August 29, 2026

Quick answer

Write three read-only Kubernetes tools behind an MCP server, poke at them with the MCP Inspector, then deploy the server in-cluster with its own ServiceAccount so RBAC — not your code — becomes the permission boundary.

intermediate · 60 min

Before you begin

  • Python 3.10+ and pip
  • A cluster you can use (kind or minikube is fine) and kubectl configured
  • Docker, for the deploy step
  • Basic familiarity with Kubernetes objects (Pods, Deployments, RBAC)
MCP
Kubernetes
AI Agents
Python
ServiceAccount
Platform Engineering

Most "connect an LLM to Kubernetes" write-ups stop at a Python function with a docstring, called directly from an agent script. That's fine for a demo and wrong for anything reused — the tool only exists inside that one script, and the credential it runs with is whatever kubeconfig happens to be on the developer's laptop.

This tutorial builds the same three tools — list pods, read logs, read events — behind an actual MCP server, so any MCP-aware client can use them, and deploys that server in-cluster with its own ServiceAccount. By the end, the permission boundary is enforced by the Kubernetes API server, not by a comment saying "this function is read-only."

What You'll Build

  • An MCP server (server.py) exposing get_pods, get_logs, and get_events as typed tools
  • A local test loop using the MCP Inspector
  • A container image and Deployment running the server in-cluster, bound to a read-only ServiceAccount
  • A Claude Code (or any MCP client) config pointing at both the local and deployed server

Step 1: Set Up the Project

bash
mkdir k8s-mcp-server && cd k8s-mcp-server
python3 -m venv .venv && source .venv/bin/activate
pip install "mcp[cli]" kubernetes

Confirm you have a cluster to point this at:

bash
kubectl cluster-info
# if you don't have one: kind create cluster --name mcp-demo

Step 2: Write the Server

Create server.py. The Python SDK reads your function's type hints to generate the tool's schema — no hand-written JSON Schema, no drift between what you documented and what the function actually accepts.

python
1# server.py
2from datetime import datetime, timezone
3
4from kubernetes import client, config
5from mcp.server import MCPServer
6
7try:
8    config.load_incluster_config()
9except config.ConfigException:
10    config.load_kube_config()
11
12core = client.CoreV1Api()
13mcp = MCPServer("kubernetes-readonly")
14
15MAX_LOG_BYTES = 20_000
16UNDATED = datetime.min.replace(tzinfo=timezone.utc)
17
18
19@mcp.tool()
20def get_pods(namespace: str, selector: str | None = None) -> str:
21    """List pods with phase, restart count, and failure reason.
22
23    Start here when you don't yet know which pod is broken.
24    """
25    pods = core.list_namespaced_pod(namespace, label_selector=selector or "")
26    if not pods.items:
27        return f"No pods found in {namespace}"
28
29    lines = []
30    for p in pods.items:
31        statuses = p.status.container_statuses or []
32        restarts = sum(c.restart_count for c in statuses)
33        reasons = [
34            c.state.waiting.reason
35            for c in statuses
36            if c.state and c.state.waiting and c.state.waiting.reason
37        ]
38        lines.append(
39            f"{p.metadata.name}  phase={p.status.phase}  restarts={restarts}"
40            + (f"  {', '.join(reasons)}" if reasons else "")
41        )
42    return "\n".join(lines)
43
44
45@mcp.tool()
46def get_logs(namespace: str, name: str, previous: bool = False) -> str:
47    """Read container logs. Set previous=true to read the container that just
48    crashed — for a restart loop, that is where the error is.
49    """
50    try:
51        text = core.read_namespaced_pod_log(
52            name=name, namespace=namespace, previous=previous, tail_lines=200
53        )
54    except client.exceptions.ApiException as e:
55        return f"Could not read logs: {e.reason}"
56
57    if len(text) > MAX_LOG_BYTES:
58        text = "...[truncated]...\n" + text[-MAX_LOG_BYTES:]
59    return text or "(empty)"
60
61
62@mcp.tool()
63def get_events(namespace: str, name: str | None = None) -> str:
64    """Recent events for a namespace or a single object. Events explain
65    scheduling failures, image pull errors, and OOM kills.
66    """
67    field = f"involvedObject.name={name}" if name else ""
68    events = core.list_namespaced_event(namespace, field_selector=field)
69    rows = sorted(events.items, key=lambda e: e.last_timestamp or e.event_time or UNDATED)
70    return "\n".join(
71        f"{e.last_timestamp}  {e.type}  {e.reason}: {e.message}" for e in rows[-25:]
72    ) or "No events found."

Two details worth noticing before you move on:

  • The docstring is the tool description a model reads. "Set previous=true … that is where the error is" is an instruction for the agent, not a comment for you.
  • get_logs truncates. Do this in the server, not the client — every consumer of this tool inherits the limit for free.

Step 3: Run It and Poke at the Tools

Launch the MCP Inspector, a browser UI for calling your tools by hand before any model is involved:

bash
mcp dev server.py

Open the URL it prints. In the Inspector:

  1. Call get_pods with namespace=kube-system — you should get a list of system pods with phase and restart counts.
  2. Pick a pod name from that list and call get_logs with it.
  3. Call get_events with the same namespace and confirm you see recent scheduling events.

If any call errors, check the exception message first — most failures at this stage are RBAC (the identity mcp dev is running as can't list pods) rather than a bug in the tool code.

Step 4: Choose a Transport

You've been running over stdio — the Inspector launched your script as a subprocess and talked over stdin/stdout. That's right for local tooling: no network, no auth to design, and the server inherits whatever kubeconfig the launching user has.

For anything shared, switch to Streamable HTTP:

bash
mcp run server.py --transport streamable-http

This binds 127.0.0.1:8000 by default, which is fine on a laptop and useless in a container. For the deployed build in the next step, bind explicitly instead of using the CLI — add this to the bottom of server.py, after the tool definitions from Step 2:

python
if __name__ == "__main__":
    mcp.run(transport="streamable-http", host="0.0.0.0", port=8000)

Without this block, server.py only registers the tools and exits — the Dockerfile's CMD in the next step needs it to actually bind and serve.

The transport choice is really an identity choice: over stdio the server runs as you; deployed over HTTP it runs as itself, with its own credential. That's what the rest of this tutorial sets up.

Step 5: Containerize It

dockerfile
1# Dockerfile
2FROM python:3.12-slim
3WORKDIR /app
4COPY server.py .
5RUN pip install --no-cache-dir "mcp[cli]" kubernetes
6USER 1000
7CMD ["python", "server.py"]
bash
docker build -t k8s-mcp-server:v0.1.0 .
# push it wherever your cluster can pull from, e.g.:
# docker tag k8s-mcp-server:v0.1.0 ghcr.io/your-org/k8s-mcp-server:v0.1.0
# docker push ghcr.io/your-org/k8s-mcp-server:v0.1.0

Step 6: Give It a ServiceAccount and Deploy In-Cluster

This is the step that actually changes the security story. Instead of a kubeconfig on a laptop, the server runs as a ServiceAccount Kubernetes controls directly.

bash
kubectl create namespace mcp
kubectl create serviceaccount k8s-mcp-server -n mcp
yaml
1# rbac.yaml
2apiVersion: rbac.authorization.k8s.io/v1
3kind: ClusterRole
4metadata:
5  name: mcp-readonly
6rules:
7  - apiGroups: [""]
8    resources: ["pods", "events"]
9    verbs: ["get", "list"]
10  - apiGroups: [""]
11    resources: ["pods/log"]
12    verbs: ["get"]
13---
14apiVersion: rbac.authorization.k8s.io/v1
15kind: ClusterRoleBinding
16metadata:
17  name: mcp-readonly-binding
18subjects:
19  - kind: ServiceAccount
20    name: k8s-mcp-server
21    namespace: mcp
22roleRef:
23  kind: ClusterRole
24  name: mcp-readonly
25  apiGroup: rbac.authorization.k8s.io
bash
kubectl apply -f rbac.yaml

Note what's deliberately absent: no secrets, no configmaps, no list on cluster-wide resources beyond what the three tools need. If you want the reasoning behind excluding those two resources specifically, see Give an AI Agent Read-Only Access to Kubernetes — this ClusterRole follows that guidance.

yaml
1# deploy.yaml
2apiVersion: apps/v1
3kind: Deployment
4metadata:
5  name: k8s-mcp-server
6  namespace: mcp
7spec:
8  replicas: 2
9  selector:
10    matchLabels: { app: k8s-mcp-server }
11  template:
12    metadata:
13      labels: { app: k8s-mcp-server }
14    spec:
15      serviceAccountName: k8s-mcp-server
16      containers:
17        - name: server
18          image: k8s-mcp-server:v0.1.0
19          command: ["python", "server.py"]
20          ports:
21            - containerPort: 8000
22          securityContext:
23            runAsNonRoot: true
24            readOnlyRootFilesystem: true
25            allowPrivilegeEscalation: false
26            capabilities:
27              drop: ["ALL"]
28          resources:
29            requests: { cpu: 100m, memory: 128Mi }
30            limits: { memory: 256Mi }
31---
32apiVersion: v1
33kind: Service
34metadata:
35  name: k8s-mcp-server
36  namespace: mcp
37spec:
38  selector: { app: k8s-mcp-server }
39  ports:
40    - port: 8000
41      targetPort: 8000
bash
kubectl apply -f deploy.yaml
kubectl -n mcp rollout status deployment/k8s-mcp-server

Step 7: Verify the Ceiling, Not Just the Happy Path

A tool that only tries to read is not the same as a credential that cannot write. Prove it:

bash
SA=system:serviceaccount:mcp:k8s-mcp-server

kubectl auth can-i delete pods --as=$SA -n mcp        # expect: no
kubectl auth can-i get secrets --as=$SA -n mcp        # expect: no
kubectl auth can-i get pods --as=$SA -n mcp           # expect: yes

If any of the first two come back yes, stop and fix the ClusterRole before connecting a client — this check is the actual security boundary, everything else is convenience.

Step 8: Connect a Client

For local development, point Claude Code (or another MCP client) at the script directly over stdio:

json
1{
2  "mcpServers": {
3    "kubernetes": {
4      "command": "mcp",
5      "args": ["run", "/path/to/server.py"]
6    }
7  }
8}

To reach the deployed server instead, port-forward it and configure your client for the Streamable HTTP transport:

bash
kubectl -n mcp port-forward svc/k8s-mcp-server 8000:8000

Either way, ask your client something like "which pods in kube-system have restarted more than twice?" and confirm it calls get_pods, not that it guesses from training data.

Where to Go Next

You now have three tools, a container, and a permission ceiling the cluster enforces rather than one your code merely claims. From here:

We built Podscape to simplify Kubernetes workflows like this — logs, events, and cluster state in one interface, without switching tools.

Struggling with this in production?

We help teams fix these exact issues. Our engineers have deployed these patterns across production environments at scale.