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

```python
# 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

boto3-eks

AWS Boto3 EKS Integration

Quick Reference

Client Initialization