aibot88 AWS Boto3 EKS Integration
Complete patterns for managing Amazon EKS clusters using AWS Boto3 SDK, including authentication token generation for Kubernetes client integration.
Quick Reference
Client Initialization
import boto3
from typing import Optional
def get_eks_client(region_name: str = 'us-east-1',
profile_name: Optional[str] = None):
"""Initialize EKS client with optional profile"""
session = boto3.Session(
region_name=region_name,
profile_name=profile_name
)
return session.client('eks')
# Usage
eks = get_eks_client(region_name='us-west-2')
Essential Cluster Operations
# Describe cluster
cluster = eks.describe_cluster(name='my-cluster')
print(f"Status: {cluster['cluster']['status']}")
print(f"Endpoint: {cluster['cluster']['endpoint']}")
print(f"Version: {cluster['cluster']['version']}")
# List clusters
clusters = eks.list_clusters()
for cluster_name in clusters['clusters']:
print(cluster_name)
# Check cluster status
waiter = eks.get_waiter('cluster_active')
waiter.wait(name='my-cluster')
Authentication Token Generation
import base64
from botocore.signers import RequestSigner
def get_eks_token(cluster_name: str, region_name: str = 'us-east-1') -> str:
"""Generate EKS authentication token (k8s-aws-v1.*)"""
session = boto3.Session(region_name=region_name)
client = session.client('sts')
service_id = client.meta.service_model.service_id
signer = RequestSigner(
service_id,
region_name,
'sts',
'v4',
session.get_credentials(),
session.events
)
params = {
'method': 'GET',
'url': f'https://sts.{region_name}.amazonaws.com/?Action=GetCallerIdentity&Version=2011-06-15',
'body': {},
'headers': {
'x-k8s-aws-id': cluster_name
},
'context': {}
}
signed_url = signer.generate_presigned_url(
params,
region_name=region_name,
expires_in=60,
operation_name=''
)
token = 'k8s-aws-v1.' + base64.urlsafe_b64encode(
signed_url.encode('utf-8')
).decode('utf-8').rstrip('=')
return token
# Usage with Kubernetes client
from kubernetes import client as k8s_client
token = get_eks_token('my-cluster', 'us-west-2')
configuration = k8s_client.Configuration()
configuration.host = cluster['cluster']['endpoint']
configuration.api_key = {"authorization": f"Bearer {token}"}
configuration.ssl_ca_cert = '/tmp/ca.crt' # Cluster CA certificate
Kubeconfig Generation
import yaml
from pathlib import Path
def generate_kubeconfig(cluster_name: str,
region_name: str,
output_path: str = '~/.kube/config') -> dict:
"""Generate kubeconfig for EKS cluster"""
eks = get_eks_client(region_name)
cluster_info = eks.describe_cluster(name=cluster_name)['cluster']
kubeconfig = {
'apiVersion': 'v1',
'kind': 'Config',
'clusters': [{
'cluster': {
'certificate-authority-data': cluster_info['certificateAuthority']['data'],
'server': cluster_info['endpoint']
},
'name': cluster_info['arn']
}],
'contexts': [{
'context': {
'cluster': cluster_info['arn'],
'user': cluster_info['arn']
},
'name': cluster_info['arn']
}],
'current-context': cluster_info['arn'],
'users': [{
'name': cluster_info['arn'],
'user': {
'exec': {
'apiVersion': 'client.authentication.k8s.io/v1beta1',
'command': 'aws',
'args': [
'eks',
'get-token',
'--cluster-name',
cluster_name,
'--region',
region_name
],
'env': None,
'interactiveMode': 'IfAvailable',
'provideClusterInfo': False
}
}
}]
}
# Write to file
output = Path(output_path).expanduser()
output.parent.mkdir(parents=True, exist_ok=True)
with output.open('w') as f:
yaml.dump(kubeconfig, f, default_flow_style=False)
return kubeconfig
Node Group Management
# List node groups
nodegroups = eks.list_nodegroups(clusterName='my-cluster')
for ng in nodegroups['nodegroups']:
print(ng)
# Describe node group
ng_info = eks.describe_nodegroup(
clusterName='my-cluster',
nodegroupName='my-nodegroup'
)
print(f"Status: {ng_info['nodegroup']['status']}")
print(f"Capacity: {ng_info['nodegroup']['scalingConfig']}")
# Update node group scaling
eks.update_nodegroup_config(
clusterName='my-cluster',
nodegroupName='my-nodegroup',
scalingConfig={
'minSize': 2,
'maxSize': 10,
'desiredSize': 4
}
)
Fargate Profiles
# List Fargate profiles
profiles = eks.list_fargate_profiles(clusterName='my-cluster')
# Describe Fargate profile
profile_info = eks.describe_fargate_profile(
clusterName='my-cluster',
fargateProfileName='my-fargate-profile'
)
print(f"Status: {profile_info['fargateProfile']['status']}")
print(f"Selectors: {profile_info['fargateProfile']['selectors']}")
Common Patterns
Error Handling
from botocore.exceptions import ClientError, BotoCoreError
try:
cluster = eks.describe_cluster(name='my-cluster')
except eks.exceptions.ResourceNotFoundException:
print("Cluster not found")
except ClientError as e:
error_code = e.response['Error']['Code']
if error_code == 'AccessDeniedException':
print("Insufficient permissions")
else:
print(f"AWS Error: {error_code}")
except BotoCoreError as e:
print(f"Connection error: {e}")
Async Operations with Waiters
# Wait for cluster to be active
waiter = eks.get_waiter('cluster_active')
waiter.wait(
name='my-cluster',
WaiterConfig={
'Delay': 30, # Poll every 30 seconds
'MaxAttempts': 40 # Max 20 minutes
}
)
# Wait for cluster deletion
waiter = eks.get_waiter('cluster_deleted')
waiter.wait(name='my-cluster')
# Wait for nodegroup active
waiter = eks.get_waiter('nodegroup_active')
waiter.wait(
clusterName='my-cluster',
nodegroupName='my-nodegroup'
)
IRSA (IAM Roles for Service Accounts)
# Get OIDC provider for cluster
cluster = eks.describe_cluster(name='my-cluster')['cluster']
oidc_issuer = cluster['identity']['oidc']['issuer']
print(f"OIDC Issuer: {oidc_issuer}")
# OIDC issuer URL (without https://)
oidc_provider = oidc_issuer.replace('https://', '')
# Use with IAM to create trust relationship
# (See references/auth-tokens.md for complete IRSA setup)
Progressive Disclosure
Quick Start (This File)
- Client initialization
- Essential cluster operations
- Token generation for K8s client
- Kubeconfig generation
- Basic node group operations
Detailed References
- Cluster Operations: Complete cluster lifecycle management, version updates, configuration changes, and advanced waiter patterns
- Authentication & Tokens: Token generation, refresh patterns, cross-account access, IRSA setup, and Kubernetes client integration
- Node Groups & Fargate: Managed node groups, Fargate profiles, add-ons, scaling operations, and update workflows
When to Use This Skill
Use this skill when:
- Managing EKS clusters programmatically
- Generating authentication tokens for Kubernetes Python client
- Creating or updating kubeconfig files
- Automating node group scaling
- Setting up Fargate profiles
- Implementing IRSA for pod-level permissions
- Cross-account cluster access
- Integrating EKS with Python applications
Dependencies
pip install boto3 botocore kubernetes pyyaml
Related Skills
- anthropic-expert: Claude API patterns for agent-based EKS management
- claude-cost-optimization: Cost tracking for EKS automation workflows
- railway-deployment: Container deployment patterns
Best Practices
- Credentials: Use IAM roles or profiles, never hardcode credentials
- Token Refresh: EKS tokens expire in 15 minutes, implement refresh logic
- Error Handling: Always handle ResourceNotFoundException and throttling
- Waiters: Use built-in waiters for async operations instead of polling
- Region Awareness: Always specify region explicitly
- IRSA: Use IRSA for pod-level permissions instead of node IAM roles
- Pagination: Use paginators for list operations in large environments
- Type Hints: Include type hints for better IDE support and validation