Get shell access to ECS tasks using execute-command

README.md

ecs-connect.py

Get shell access to ECS tasks using execute-command.

Local Prerequisites

1. aws CLI
2. uv
3. Install the Session Manager Plugin
4. AWS Credentials

ECS Prerequisites

1. ECS Task Role has the necessary permissions
2. ECS Task started with --enable-execute-command

Installation

chmod +x ecs-connect.py

Usage

./ecs-connect.py

It will prompt you to choose a cluster and task to connect to.

ecs-connect.py

#!/usr/bin/env -S uv run --script

# /// script
# requires-python = ">=3.12"
# dependencies = [
#   "boto3",
#   "rich",
#   "inquirer",
# ]
# ///

import os
import sys

import boto3
import inquirer
from rich import print as rprint
from rich.console import Console


def get_clusters(ecs):
    clusters = []
    paginator = ecs.get_paginator("list_clusters")

    for page in paginator.paginate():
        clusters.extend([arn.split("/")[-1] for arn in page["clusterArns"]])

    return clusters


def get_tasks(ecs, cluster):
    tasks = {}
    paginator = ecs.get_paginator("list_tasks")

    for page in paginator.paginate(cluster=cluster):
        if page["taskArns"]:
            task_details = ecs.describe_tasks(cluster=cluster, tasks=page["taskArns"])
            for task in task_details["tasks"]:
                task_id = task["taskArn"].split("/")[-1]
                task_definition = task["taskDefinitionArn"].split("/")[-1]
                containers = task["containers"]
                running_containers = [
                    c for c in containers if c["lastStatus"] == "RUNNING"
                ]

                tasks[task_id] = {
                    "label": f"{task_definition} ({task_id})",
                    "arn": task["taskArn"],
                    "containers": containers,
                    "running_containers": running_containers,
                }

    return tasks


def select_container(containers):
    if not containers:
        rprint("[red]No running containers found in the task[/red]")
        sys.exit(1)

    if len(containers) == 1:
        return containers[0]

    container_choices = [(c["name"], c) for c in containers]
    container_question = [
        inquirer.List(
            "container",
            message="Select container to connect to",
            choices=container_choices,
        )
    ]

    container_answer = inquirer.prompt(container_question)
    if not container_answer:
        sys.exit(1)

    return container_answer["container"]


def main():
    console = Console()

    try:
        ecs = boto3.client("ecs")
        session = boto3.Session()
        region = session.region_name

        if not region:
            rprint(
                "[red]No AWS region configured. Please set AWS_REGION or configure your AWS CLI.[/red]"
            )
            sys.exit(1)

        clusters = get_clusters(ecs)

        if not clusters:
            rprint(f"[red]No ECS clusters found in region {region}[/red]")
            sys.exit(1)

        questions = [
            inquirer.List("cluster", message="Select ECS cluster", choices=clusters),
        ]

        answers = inquirer.prompt(questions)
        if not answers:
            sys.exit(1)

        cluster = answers["cluster"]
        tasks = get_tasks(ecs, cluster)

        if not tasks:
            rprint(f"[red]No tasks found in cluster {cluster}[/red]")
            sys.exit(1)

        task_choices = [(v["label"], k) for k, v in tasks.items()]
        task_question = [
            inquirer.List(
                "task",
                message="Select task to connect to",
                choices=task_choices,
            )
        ]

        task_answer = inquirer.prompt(task_question)
        if not task_answer:
            sys.exit(1)

        task = tasks[task_answer["task"]]

        # Handle container selection
        selected_container = select_container(task["running_containers"])
        container_name = selected_container["name"]

        rprint(
            f"[green]Connecting to container {container_name} in task {task['label']} in cluster {cluster}...[/green]"
        )

        os.execvp(
            "aws",
            [
                "aws",
                "ecs",
                "execute-command",
                "--cluster",
                cluster,
                "--task",
                task["arn"],
                "--container",
                container_name,
                "--command",
                "/bin/bash",
                "--interactive",
            ],
        )

    except Exception as e:
        rprint(f"[red]Error: {str(e)}[/red]")
        sys.exit(1)


if __name__ == "__main__":
    main()

(credit to Claude for writing most of this)