• October 21, 2026
• 11 min read

Amazon ECS — boto3 in Python¶

ECS is a JSON-protocol API with 88 operations in boto3 1.43.116 and no resource-style interface: use the ecs client (boto3.client("ecs")). Sessions, Config, retry modes, pagination basics and ClientError are explained in AWS Foundations; this page covers what is specific to ECS and the traps that make test code flaky.

import boto3
from botocore.config import Config

ecs = boto3.client("ecs", config=Config(retries={"mode": "standard", "max_attempts": 5}, connect_timeout=5, read_timeout=30))

Modeled exceptions hang off the client: ecs.exceptions.ClusterNotFoundException, ecs.exceptions.ServiceNotFoundException, ecs.exceptions.InvalidParameterException.

Operations You Will Use¶

Task Definitions and Clusters¶

from app.taskdef import build_task_definition                     # builder from 01

td = build_task_definition("orders-api", "123456789012.dkr.ecr.us-east-1.amazonaws.com/orders:1.4.2",
                           execution_role_arn=EXEC_ROLE, task_role_arn=TASK_ROLE, port=8000)
reg = ecs.register_task_definition(**td)["taskDefinition"]
print(reg["taskDefinitionArn"], reg["revision"], reg["status"])

arn:aws:ecs:us-east-1:123456789012:task-definition/orders-api:1 1 ACTIVE

• Registering is not idempotent: every call creates a new revision, even for identical JSON. In CI, register once per image tag and pass the returned ARN on.
• Use the revision-qualified name (orders-api:8) or ARN when running or deploying. A bare family means "latest ACTIVE revision", which can change between your describe and your run.
• IAM policies that name task definitions use the ARN with a revision or :* for all revisions.

ecs.create_cluster(
    clusterName="prod",
    capacityProviders=["FARGATE", "FARGATE_SPOT"],
    defaultCapacityProviderStrategy=[{"capacityProvider": "FARGATE", "weight": 1}],   # used when run_task names neither launchType nor strategy
    settings=[{"name": "containerInsights", "value": "enabled"}],
    tags=[{"key": "env", "value": "prod"}],
)

Running a Task¶

run_task takes the cluster, a revision, and where and how to run it:

One reusable function¶

app/runner.py is the heart of every test harness and CI job in this guide. It makes the four failure modes visible: placement failures, a task that never stops, a non-zero exit code, and a container that never started.

class TaskStartError(RuntimeError):
    """run_task answered HTTP 200 but placed no task: the reasons are in `failures`."""
    def __init__(self, failures):
        self.failures = failures
        super().__init__("; ".join(f"{f.get('reason')} ({f.get('detail') or f.get('arn')})" for f in failures))

@dataclass
class TaskResult:
    task_arn: str; stop_code: str | None; stopped_reason: str | None
    exit_codes: dict[str, int | None]; container_reasons: dict[str, str | None]
    @property
    def ok(self) -> bool:
        return bool(self.exit_codes) and all(code == 0 for code in self.exit_codes.values())   # None is NOT success

def run_and_wait(ecs, *, cluster, task_definition, subnets, security_groups, command=None, environment=None, container="app",
                 started_by="qa-harness", tags=None, capacity_provider="FARGATE", assign_public_ip="DISABLED",
                 timeout_s=600, poll_s=6) -> TaskResult:
    overrides = {}
    if command or environment:
        override = {"name": container}
        if command:
            override["command"] = command
        if environment:
            override["environment"] = [{"name": k, "value": v} for k, v in environment.items()]
        overrides = {"containerOverrides": [override]}

    resp = ecs.run_task(
        cluster=cluster, taskDefinition=task_definition, count=1,
        capacityProviderStrategy=[{"capacityProvider": capacity_provider, "weight": 1}],
        networkConfiguration={"awsvpcConfiguration": {"subnets": subnets, "securityGroups": security_groups, "assignPublicIp": assign_public_ip}},
        overrides=overrides, startedBy=started_by, tags=[{"key": k, "value": v} for k, v in (tags or {}).items()],
        propagateTags="TASK_DEFINITION",
        clientToken=uuid.uuid4().hex,                                     # stays the same across botocore's own retries of this call
    )
    if resp["failures"] or not resp["tasks"]:                             # HTTP 200 does not mean "started"
        raise TaskStartError(resp["failures"])
    task_arn = resp["tasks"][0]["taskArn"]
    try:
        ecs.get_waiter("tasks_stopped").wait(cluster=cluster, tasks=[task_arn],
                                             WaiterConfig={"Delay": poll_s, "MaxAttempts": max(1, timeout_s // max(poll_s, 1))})
    except WaiterError:
        ecs.stop_task(cluster=cluster, task=task_arn, reason="harness timeout")      # never leave a billed task behind
        raise TimeoutError(f"task {task_arn} still running after {timeout_s}s; stop requested") from None

    task = ecs.describe_tasks(cluster=cluster, tasks=[task_arn])["tasks"][0]
    result = TaskResult(task_arn, task.get("stopCode"), task.get("stoppedReason"),
                        {c["name"]: c.get("exitCode") for c in task["containers"]},
                        {c["name"]: c.get("reason") for c in task["containers"]})
    if not result.ok:
        raise TaskFailedError(result)
    return result

(Excerpt: the real file also checks failures on the final describe_tasks, and TaskFailedError builds its message from the result.)

result = run_and_wait(ecs, cluster="qa", task_definition="migrate:4", subnets=[subnet], security_groups=[group],
                      command=["python", "-m", "migrate", "--to", "head"], environment={"STAGE": "test"}, tags={"suite": "pytest"})
print(result.exit_codes)            # {'app': 0}

Waiters¶

Always pass WaiterConfig={"Delay": ..., "MaxAttempts": ...}: the defaults are fine for a deploy and far too slow for a unit test (Delay: 0 against moto). On timeout the waiter raises botocore.exceptions.WaiterError; a tasks_running wait on a task that crashes raises it immediately with the state in the message, which is useful: it fails fast instead of waiting 10 minutes. tasks_stopped does not tell you whether the stop was good; read the exit codes afterwards. services_stable is true after an automatic rollback too (02).

Detecting Failure¶

No single field answers "did it work?". Read them together, in this order:

Two traps: exitCode can be missing, so all(c["exitCode"] == 0 ...) is wrong where c.get("exitCode") == 0 is right; and a task with several containers can have a zero-exit essential container and a failing sidecar, so check each container you care about.

api_failures_messages in the AWS guide lists the failures[].reason values: MISSING, and for run_task RESOURCE:CPU, RESOURCE:MEMORY, RESOURCE:ENI, RESOURCE:GPU, AGENT, LOCATION, ATTRIBUTE (mostly relevant to EC2 capacity).

HTTP 200 with failures¶

resp = ecs.describe_services(cluster="qa", services=["ghost"])
print(resp["ResponseMetadata"]["HTTPStatusCode"], resp["services"], [f["reason"] for f in resp["failures"]])

200 [] ['MISSING']

describe_* and run_task/start_task report per-resource problems in failures and do not raise. Other operations raise: update_service on an unknown service raises ServiceNotFoundException, list_tasks on an unknown cluster raises ClusterNotFoundException. Wrap both styles in one helper, as run_and_wait does, and test both (05).

Services: Update, Roll Out, Stop¶

ecs.update_service(cluster="prod", service="orders-api", taskDefinition="orders-api:9")                  # new revision: rolling deployment
ecs.update_service(cluster="prod", service="orders-api", forceNewDeployment=True)                        # same revision: replace tasks (new image behind the same tag, new secrets, new platform revision)
ecs.update_service(cluster="prod", service="orders-api", desiredCount=0)                                 # required before delete_service
ecs.delete_service(cluster="prod", service="orders-api")                                                  # service goes DRAINING, then INACTIVE
ecs.stop_task(cluster="prod", task=task_arn, reason="qa cleanup")                                         # SIGTERM, then SIGKILL after stopTimeout

• update_service with a task definition, desired count or network change starts a deployment; with no change it does nothing unless forceNewDeployment=True.
• The documented rule for delete_service: scale to 0 first (or pass force=True); the service then drains and becomes INACTIVE. Use the services_inactive waiter before re-creating a service of the same name.
• stop_task is asynchronous: it records desiredStatus: STOPPED and returns while the containers are still shutting down (SIGTERM, then SIGKILL after stopTimeout). Wait for tasks_stopped if the next step depends on it. A stopped service task is replaced by its service; stop the service's capacity instead (desiredCount) if you want it gone.

Pagination and Batch Limits¶

def chunks(items, size):
    for i in range(0, len(items), size):
        yield items[i:i + size]

task_arns = [a for page in ecs.get_paginator("list_tasks").paginate(cluster="qa", startedBy="qa-harness") for a in page["taskArns"]]
tasks = [t for batch in chunks(task_arns, 100) for t in ecs.describe_tasks(cluster="qa", tasks=batch)["tasks"]]

Without a paginator, a plain list_services call returns only 10 services and a nextToken. list_tasks shows only RUNNING tasks by default; pass desiredStatus="STOPPED" for finished ones. list_* return identifiers; call describe_* for details. moto ignores maxResults on list_clusters, list_tasks and list_services (06), so a pagination bug will not show in moto tests.

Errors¶

Throttling is per account and Region, with a token bucket per operation category: DescribeTasks and ListTasks allow a burst of 100 and 20 per second sustained, RunTask and StopTask 100 burst and 40 per second, UpdateService 50 burst and 5 per second. A poll loop that describes many tasks every second across parallel tests hits it. Poll less often, batch IDs in one describe_tasks, use EventBridge events, and keep retries.mode=standard.

Idempotency and Retries¶

• run_task(clientToken=...) (up to 64 characters) and create_service(clientToken=...) (up to 36) make a retried request safe. botocore resends the same parameters when it retries, so a token generated before the call protects against its automatic retries; generate one per logical job, not per attempt of your own loop.
• register_task_definition has no token: a retry can create an extra revision. Harmless, but do not use revision numbers as counters.
• update_service to the same state is a no-op; stop_task on a stopped task returns the task.
• Never loop run_task on failures blindly: RESOURCE:* and AGENT mean capacity is missing, and a loop turns that into a throttling problem. Back off, or fail the test.
• stop_task in a finally and a janitor for leftovers (list by startedBy/tag, stop, deregister) keep a failed test run from costing money.

Running a Task from a pytest Harness¶

# tests/smoke/test_job.py (shape; the real fixtures are in 05 and 06)
@pytest.mark.aws_smoke
def test_migration_exits_zero(ecs, cfg, taskdef):
    result = run_and_wait(ecs, cluster=cfg["cluster"], task_definition=taskdef, subnets=cfg["subnets"],
                          security_groups=cfg["security_groups"], command=["python", "-m", "migrate"], tags={"purpose": "smoke"})
    assert result.ok

Failures then read as TaskFailedError: task 1f3a... failed: stopCode=EssentialContainerExited reason='Essential container in task exited' exit={'app': 3} instead of a bare timeout.

Quick Rules¶

1. Use the ecs client, qualify task definitions as family:revision, and register once per image tag.
2. After run_task, check failures before reading tasks. After describe_*, check failures for MISSING.
3. Pass WaiterConfig and decide per waiter what failure means; tasks_stopped never fails early.
4. Treat a missing exitCode as failure and read stopCode, stoppedReason and containers[].reason with it.
5. Always stop what you start: stop_task in finally, tags and startedBy for the janitor.
6. Generate clientToken once per logical call so botocore retries cannot duplicate tasks.
7. Batch and page: 100 tasks and 10 services per describe, paginators for list_*, and do not poll in a tight loop.
8. Catch modeled exceptions (ecs.exceptions.ClusterNotFoundException), and remember which operations return failures instead of raising.
9. After services_stable, check which revision is live; it also succeeds after a rollback.

See also¶

• Amazon ECS — Overview
• Amazon ECS — Services, Deployments & Networking
• Amazon ECS — Testing ECS-Based Systems
• AWS Foundations — Clients, Config, Retries & Errors
• AWS Foundations — Credentials, IAM & Local Testing
• Pytest — Python Testing Framework