Microsoft Azure Container Service Management Client Library for Python
npx @tessl/cli install tessl/pypi-azure-mgmt-containerservice@39.1.0A comprehensive Python client library for managing Azure Kubernetes Service (AKS) resources through the Azure Resource Manager REST API. This library enables developers to programmatically create, configure, and manage Kubernetes clusters on Azure, including operations for cluster lifecycle management, node pool configuration, networking setup, and security policies.
pip install azure-mgmt-containerservicefrom azure.mgmt.containerservice import ContainerServiceClient
from azure.identity import DefaultAzureCredentialFor async operations:
from azure.mgmt.containerservice.aio import ContainerServiceClient as AsyncContainerServiceClient
from azure.identity.aio import DefaultAzureCredential as AsyncDefaultAzureCredentialfrom azure.mgmt.containerservice import ContainerServiceClient
from azure.identity import DefaultAzureCredential
# Initialize the client
credential = DefaultAzureCredential()
subscription_id = "your-subscription-id"
client = ContainerServiceClient(credential, subscription_id)
# List all managed clusters
clusters = client.managed_clusters.list()
for cluster in clusters:
print(f"Cluster: {cluster.name}, Location: {cluster.location}")
# Get a specific cluster
cluster = client.managed_clusters.get("resource-group-name", "cluster-name")
print(f"Cluster status: {cluster.provisioning_state}")
# Close the client
client.close()Async usage:
import asyncio
from azure.mgmt.containerservice.aio import ContainerServiceClient
from azure.identity.aio import DefaultAzureCredential
async def main():
credential = DefaultAzureCredential()
async with ContainerServiceClient(credential, subscription_id) as client:
# List clusters asynchronously
clusters = []
async for cluster in client.managed_clusters.list():
clusters.append(cluster)
print(f"Found {len(clusters)} clusters")
asyncio.run(main())The client follows the Azure SDK architecture pattern with a main client class that provides access to operation groups:
Both synchronous and asynchronous versions follow the same API patterns, with async methods returning coroutines.
Core AKS cluster management including creation, updates, scaling, credential rotation, and deletion. Supports both basic and advanced cluster configurations with comprehensive networking, security, and monitoring options.
def get(resource_group_name: str, resource_name: str, **kwargs) -> ManagedCluster: ...
def begin_create_or_update(resource_group_name: str, resource_name: str, parameters: ManagedCluster, **kwargs) -> LROPoller[ManagedCluster]: ...
def begin_delete(resource_group_name: str, resource_name: str, **kwargs) -> LROPoller[None]: ...
def list(**kwargs) -> ItemPaged[ManagedCluster]: ...
def list_by_resource_group(resource_group_name: str, **kwargs) -> ItemPaged[ManagedCluster]: ...Node pool management for AKS clusters including system and user node pools, auto-scaling configuration, OS and VM size selection, and maintenance operations.
def get(resource_group_name: str, resource_name: str, agent_pool_name: str, **kwargs) -> AgentPool: ...
def begin_create_or_update(resource_group_name: str, resource_name: str, agent_pool_name: str, parameters: AgentPool, **kwargs) -> LROPoller[AgentPool]: ...
def begin_delete(resource_group_name: str, resource_name: str, agent_pool_name: str, **kwargs) -> LROPoller[None]: ...
def list(resource_group_name: str, resource_name: str, **kwargs) -> ItemPaged[AgentPool]: ...Scheduled maintenance windows and maintenance configuration management for AKS clusters, including both node OS updates and Kubernetes version upgrades.
def get(resource_group_name: str, resource_name: str, config_name: str, **kwargs) -> MaintenanceConfiguration: ...
def create_or_update(resource_group_name: str, resource_name: str, config_name: str, parameters: MaintenanceConfiguration, **kwargs) -> MaintenanceConfiguration: ...
def delete(resource_group_name: str, resource_name: str, config_name: str, **kwargs) -> None: ...
def list_by_managed_cluster(resource_group_name: str, resource_name: str, **kwargs) -> ItemPaged[MaintenanceConfiguration]: ...Private endpoint connections, private link resources, and private link service ID resolution for secure, private connectivity to AKS clusters.
def get(resource_group_name: str, resource_name: str, private_endpoint_connection_name: str, **kwargs) -> PrivateEndpointConnection: ...
def update(resource_group_name: str, resource_name: str, private_endpoint_connection_name: str, parameters: PrivateEndpointConnection, **kwargs) -> PrivateEndpointConnection: ...
def begin_delete(resource_group_name: str, resource_name: str, private_endpoint_connection_name: str, **kwargs) -> LROPoller[None]: ...
def list(resource_group_name: str, resource_name: str, **kwargs) -> PrivateEndpointConnectionListResult: ...Node pool snapshot operations for backup, restore, and blue-green deployment scenarios, enabling reliable cluster state management.
def get(resource_group_name: str, resource_name: str, **kwargs) -> Snapshot: ...
def create_or_update(resource_group_name: str, resource_name: str, parameters: Snapshot, **kwargs) -> Snapshot: ...
def delete(resource_group_name: str, resource_name: str, **kwargs) -> None: ...Trusted access role bindings and role management for secure integration with other Azure services and custom applications.
def get(resource_group_name: str, resource_name: str, trusted_access_role_binding_name: str, **kwargs) -> TrustedAccessRoleBinding: ...
def begin_create_or_update(resource_group_name: str, resource_name: str, trusted_access_role_binding_name: str, trusted_access_role_binding: TrustedAccessRoleBinding, **kwargs) -> LROPoller[TrustedAccessRoleBinding]: ...
def begin_delete(resource_group_name: str, resource_name: str, trusted_access_role_binding_name: str, **kwargs) -> LROPoller[None]: ...
def list(resource_group_name: str, resource_name: str, **kwargs) -> ItemPaged[TrustedAccessRoleBinding]: ...Individual node/machine management within agent pools, including machine-specific operations, status monitoring, and maintenance actions.
def get(resource_group_name: str, resource_name: str, agent_pool_name: str, machine_name: str, **kwargs) -> Machine: ...
def list(resource_group_name: str, resource_name: str, agent_pool_name: str, **kwargs) -> ItemPaged[Machine]: ...Lists all available operations in the Azure Container Service provider for API discovery and documentation purposes.
def list(**kwargs) -> ItemPaged[OperationValue]: ...Manages private link resources for AKS managed clusters to enable private cluster functionality and secure connectivity.
def list(resource_group_name: str, resource_name: str, **kwargs) -> PrivateLinkResourcesListResult: ...Resolves private link service IDs for managed clusters, essential for establishing private connectivity configurations.
def post(resource_group_name: str, resource_name: str, parameters: PrivateLinkResource, **kwargs) -> PrivateLinkResource: ...Manages trusted access roles that can be granted to external services for secure access to AKS clusters, with regional role definitions.
def list(location: str, **kwargs) -> ItemPaged[TrustedAccessRole]: ...Essential data models representing AKS resources including cluster specifications, agent pool configurations, network profiles, and authentication settings.
class ManagedCluster: ...
class AgentPool: ...
class ContainerServiceNetworkProfile: ...
class ManagedClusterIdentity: ...The client requires Azure credentials with appropriate permissions. Common authentication patterns:
# Using DefaultAzureCredential (recommended)
from azure.identity import DefaultAzureCredential
credential = DefaultAzureCredential()
# Using service principal
from azure.identity import ClientSecretCredential
credential = ClientSecretCredential(
tenant_id="tenant-id",
client_id="client-id",
client_secret="client-secret"
)
# Using managed identity
from azure.identity import ManagedIdentityCredential
credential = ManagedIdentityCredential()The client raises standard Azure SDK exceptions:
from azure.core.exceptions import ResourceNotFoundError, HttpResponseError
try:
cluster = client.managed_clusters.get("rg", "cluster-name")
except ResourceNotFoundError:
print("Cluster not found")
except HttpResponseError as e:
print(f"HTTP error: {e.status_code} - {e.message}")