kubernetes-asyncio is an asynchronous (AsyncIO) client library for interacting with the Kubernetes API. It provides non-blocking access to Kubernetes clusters from Python applications, built using the same OpenAPI generator approach as the official Kubernetes Python client, but with full async/await support. The library is actively maintained, with releases frequently aligning with the Kubernetes project's own approximately three-times-a-year release cadence.
pip install kubernetes_asyncioVerified import paths — ran on the pinned version, not inferred.
This quickstart example demonstrates how to load Kubernetes configuration, create an API client using an async context manager, and then list all pods across all namespaces in your cluster.
Refer to the library's official documentation and release notes for specific version compatibility with Kubernetes API versions. Map the first two parts of `kubernetes_asyncio`'s version to the Kubernetes API version you are targeting.
Explicitly set the event loop policy at the start of your application: `import asyncio; asyncio.set_event_loop_policy(asyncio.WindowsProactorEventLoopPolicy())`.
Always initialize and use `ApiClient` objects like this: `async with ApiClient() as api: ...`
Ensure your Kubernetes configuration file (e.g., `~/.kube/config`) is valid and contains the 'current-context' key. You can generate a valid config using `kubectl config view --raw` or ensure your environment variables (like `KUBECONFIG`) point to a correctly configured file.
Ensure a valid Kubernetes configuration file exists and is accessible. This file must contain at least a `current-context` key. You can explicitly specify the config file path using `config.load_kube_config(config_file='path/to/kubeconfig')` or ensure it's in a default location like `~/.kube/config` or referenced by the `KUBECONFIG` environment variable.
Install the library using pip: `pip install kubernetes-asyncio`
Ensure `config.load_kube_config()` or `config.load_incluster_config()` is correctly called to load credentials, and verify that the associated Kubernetes user or service account has the required RBAC permissions for the API operations you are attempting.
Always check if the result of an API call or an object in the returned list is `None` before attempting to access its attributes. For example: `if ret and ret.items: for i in ret.items: if i.status: print(i.status.pod_ip)`
Prepend the call to the `async` function with `await`. For example, change `v1.list_pod_for_all_namespaces()` to `await v1.list_pod_for_all_namespaces()`. Ensure your code is running within an `asyncio` event loop (e.g., `asyncio.run(main())`).
Ensure that `config.load_kube_config()` (or `config.load_incluster_config()`) is successfully awaited and returns a valid `ApiClient` or `Configuration` object before passing it to `DynamicClient`. For example: `await config.load_kube_config(); client = kubernetes_asyncio.client.ApiClient(); dyn_client = DynamicClient(client)`