Kubernetes Python Client 操作指南

使用官方 kubernetes Python 套件操作 K8s API：kubeconfig 載入、遠端連線、RBAC 設定、SSL 驗證與 Watch 事件監聽。

概述

kubernetes 是 Kubernetes 的官方 Python Client 套件，提供對所有 K8s API 的 Python 封裝，支援從本機或遠端管理叢集資源。與直接呼叫 REST API 相比，Python Client 提供型別提示、錯誤處理和物件模型，更適合用於自動化腳本和控制器開發。

安裝方式：

pip install kubernetes

• GitHub：kubernetes-client/python
• API 文件：API Endpoints 完整列表

核心內容

常用 API 類別

API 類別	用途
AppsV1Api	Deployments、StatefulSets、DaemonSets、ReplicaSets
CoreV1Api	Pods、Services、ConfigMaps、Secrets、Namespaces、PVCs
BatchV1Api	Jobs、CronJobs
NetworkingV1Api	Ingress、NetworkPolicy
AutoscalingV2Api	HorizontalPodAutoscaler

CoreV1Api 文件參考：CoreV1Api HTML 文件

載入 Kubeconfig

預設路徑（~/.kube/config）：

from kubernetes import client, config

config.load_kube_config()

指定非預設路徑（如 k3s 環境）：

config.load_kube_config("/etc/rancher/k3s/k3s.yaml")

在 Pod 內部執行（使用 in-cluster config）：

config.load_incluster_config()

遠端存取 API（無法直接操作 Master Node）

當腳本執行環境無法直接存取 K8s API Server 時，有兩種方式繞過：

方法一：透過 kubectl proxy

在叢集節點上啟動 Proxy，開放所有來源存取：

kubectl proxy --address='0.0.0.0' --port=8080 --accept-hosts='^*$'

測試連線：

curl http://localhost:8080/api/

Python Client 指向 proxy 位址：

configuration = client.Configuration()
configuration.host = "http://<node-ip>:8080"

kubectl proxy 本身負責驗證，腳本端無需提供 Token 或憑證。

方法二：Bearer Token 直接連線

取得 API Server 位址與 Token：

APISERVER=$(kubectl config view --minify | grep server | cut -f 2- -d ":" | tr -d " ")
SECRET_NAME=$(kubectl get secrets | grep ^default | cut -f1 -d ' ')
TOKEN=$(kubectl describe secret $SECRET_NAME | grep -E '^token' | cut -f2 -d':' | tr -d " ")

AKS 注意： AKS 預設沒有 Secret Token，改用以下方式取得：

kubectl config view --raw -o jsonpath='{.users[0].user.token}'

測試 API 連線：

curl $APISERVER/api --header "Authorization: Bearer $TOKEN" --insecure

Python Client 使用 Bearer Token：

configuration = client.Configuration()
configuration.host = APISERVER
configuration.verify_ssl = False
configuration.api_key = {"authorization": "Bearer " + TOKEN}

服務帳號權限設定

即使連線測試成功，Python Client 呼叫 API 仍可能因預設帳號（default ServiceAccount）權限不足而報錯 403 Forbidden。需建立具有足夠權限的 ServiceAccount 並綁定 ClusterRole：

建立 ServiceAccount 並綁定 cluster-admin：

apiVersion: v1
kind: ServiceAccount
metadata:
  name: tos-api
  namespace: default
---
kind: ClusterRoleBinding
apiVersion: rbac.authorization.k8s.io/v1
metadata:
  name: tos-api
subjects:
  - kind: ServiceAccount
    name: tos-api
    namespace: default
roleRef:
  kind: ClusterRole
  name: cluster-admin
  apiGroup: rbac.authorization.k8s.io

套用後取得 Token：

kubectl create token tos-api -n default

生產環境應依最小權限原則，建立自訂 ClusterRole 或 Role，只授予腳本實際需要的資源與動作。

SSL 憑證驗證

若不想使用 verify_ssl = False（跳過憑證驗證），可提供 CA 憑證進行正確的 SSL 驗證。

方式一：從現有 kubeconfig 匯出 Server CA 憑證：

kubectl config view --raw \
  -o jsonpath='{.clusters[?(@.name == "wyhqoneqaks")].cluster.certificate-authority-data}' \
  | base64 -d > server-ca.crt

將叢集名稱（wyhqoneqaks）替換為實際的 context name。

方式二：k3s 自簽憑證替換流程：

1. 停止 k3s 服務
2. 清除憑證目錄：/var/lib/rancher/k3s/server/tls
3. 將自產的 server-ca.crt 放入該目錄
4. 重新啟動 k3s，即可用自有 CA 簽發後續憑證

若只需 API Server 驗證，只需 server-ca.crt；完整自管所有憑證則需產生三種憑證（參考 k3s Issue）。

在 Python Client 中使用 CA 憑證：

configuration = client.Configuration()
configuration.ssl_ca_cert = "/path/to/server-ca.crt"
configuration.verify_ssl = True

Watch 功能：監聽物件變更

Watch 是 K8s API 的長連線機制，讓腳本能即時接收 Pod 新增、更新、刪除等事件，不需輪詢。

基本用法：

from kubernetes import client, config, watch

config.load_kube_config()

k8s_core_v1 = client.CoreV1Api()
k8s_watch = watch.Watch()

for event in k8s_watch.stream(
    k8s_core_v1.list_namespaced_pod,
    namespace="default",
    timeout_seconds=60
):
    event_type = event["type"]   # ADDED / MODIFIED / DELETED
    pod_name = event["object"].metadata.name
    print(f"{event_type}: {pod_name}")

處理自動斷線問題（10 分鐘斷線）：

Watch 連線約 10 分鐘後，kubelet 會主動關閉連線並拋出 ProtocolError。解決方式是捕捉錯誤並透過 resource_version 斷線重連後從中斷處繼續：

from urllib3.exceptions import ProtocolError

def get_resource_version():
    pod_list = k8s_core_v1.list_namespaced_pod(namespace="default")
    return pod_list.metadata.resource_version

resource_version = get_resource_version()

while True:
    try:
        for event in k8s_watch.stream(
            k8s_core_v1.list_namespaced_pod,
            namespace="default",
            timeout_seconds=500,
            resource_version=resource_version
        ):
            # 處理事件
            resource_version = event["object"].metadata.resource_version
    except ProtocolError:
        print("kubelet 斷線，從 resource_version 重連...")
        resource_version = get_resource_version()

resource_version 是 K8s 的事件游標，確保重連後不會遺漏或重複處理事件。

關鍵要點

• CoreV1Api 處理 Pod/Service/ConfigMap 等核心資源；AppsV1Api 處理 Deployment/StatefulSet
• 預設 default ServiceAccount 權限不足，需建立專用 ServiceAccount 並綁定適當 ClusterRole
• Watch 連線約 10 分鐘後 kubelet 會斷線，必須捕捉 ProtocolError 並用 resource_version 重連
• AKS 無預設 Secret Token，需用 kubectl config view --raw 或 kubectl create token 取得
• 生產環境應提供 CA 憑證（ssl_ca_cert）而非跳過 SSL 驗證

實際應用

Python Client 常用於以下場景：

• 自動化維運腳本：批次更新多個 Deployment 的環境變數或映像版本
• 自訂控制器（Controller）：用 Watch 監聽 CRD 或 Pod 狀態變更，觸發自動化動作
• 叢集健康監測：定期掃描所有 Namespace 的 Pod 狀態，發送告警

部署設定參考

以下為實際使用的設定與程式碼片段，供日後查詢與複製使用。

ServiceAccount YAML

apiVersion: v1
kind: ServiceAccount
metadata:
  name: tos-api
  namespace: default
---
kind: ClusterRoleBinding
apiVersion: rbac.authorization.k8s.io/v1
metadata:
  name: tos-api
subjects:
  - kind: ServiceAccount
    name: tos-api
    namespace: default
roleRef:
  kind: ClusterRole
  name: cluster-admin
  apiGroup: rbac.authorization.k8s.io

操作指令

# 啟動 kubectl proxy（允許所有來源連線）
kubectl proxy --address='0.0.0.0' --port=8080 --accept-hosts='^*$'

# 取得 API Server 位址與 Token（非 AKS 環境）
APISERVER=$(kubectl config view --minify | grep server | cut -f 2- -d ":" | tr -d " ")
SECRET_NAME=$(kubectl get secrets | grep ^default | cut -f1 -d ' ')
TOKEN=$(kubectl describe secret $SECRET_NAME | grep -E '^token' | cut -f2 -d':' | tr -d " ")

# 取得 Token（AKS 環境）
kubectl config view --raw -o jsonpath='{.users[0].user.token}'

# 建立 ServiceAccount Token（K8s 1.24+）
kubectl create token tos-api -n default

# 匯出 Server CA 憑證
kubectl config view --raw \
  -o jsonpath='{.clusters[?(@.name == "wyhqoneqaks")].cluster.certificate-authority-data}' \
  | base64 -d > server-ca.crt

相關概念

• Kubernetes RBAC 帳號管理 — ServiceAccount 與 ClusterRoleBinding 的完整設計
• OpenSSL 自簽憑證產生 — 為 k3s 或自建叢集產生 CA 憑證
• Kubernetes 維運技巧速查 — kubeconfig 管理、kubectl 常用操作
• KEDA（Kubernetes Event-Driven Autoscaling） — Watch 機制的高階應用：事件驅動擴縮

來源

• kubernetes-client/python GitHub
• K8s API 遠端存取文件
• Watch 斷線問題 Issue
• Python Watch timeout 設定說明
• 原始素材：2026-04-12 Kubernetes Python Client 操作指南