Great question! Do you want to: - run your flow inside an image that lives in ECR (as a Prefect deployment), or - start a separate ECR-based container from within a flow? Also, are you targeting Docker workers (pull on a host) or AWS ECS workers (Fargate/EC2)? Below are quick recipes for both. Option A — Run the flow in an ECR image (Prefect 3.x) Docker worker (pulls from ECR on a host) - Prereqs: install extras

pip install 'prefect[docker]'

, make sure the host running the Docker worker can pull from your ECR - Easiest auth: log in the worker host to ECR once and let Docker pull

aws ecr get-login-password --region us-east-1 \
| docker login --username AWS --password-stdin <http://123456789012.dkr.ecr.us-east-1.amazonaws.com|123456789012.dkr.ecr.us-east-1.amazonaws.com>

- Deploy your flow to a Docker work pool using the existing ECR image

from prefect import flow
from prefect_docker import DockerRegistryCredentials

@flow(log_prints=True)
def my_flow():
    print("Hello from ECR image")
    return "ok"

if __name__ == "__main__":
    # Optional: use a Prefect block for registry creds instead of docker login on the host
    # ecr_creds = DockerRegistryCredentials.load("my-ecr-creds")
    my_flow.deploy(
        name="my-ecr-deployment",
        work_pool_name="my-docker-pool",
        image="<http://123456789012.dkr.ecr.us-east-1.amazonaws.com/my-repo:latest|123456789012.dkr.ecr.us-east-1.amazonaws.com/my-repo:latest>",
        build=False,
        push=False,
        # job_variables={"registry_credentials": ecr_creds},  # uncomment if using a block
    )

- Create pool and start worker (verified for Prefect 3.x)

prefect work-pool create --type docker my-docker-pool
prefect worker start --pool my-docker-pool --type docker

ECS worker (ECR pull handled by ECS execution role) - Prereqs: install extras

pip install 'prefect[aws]'

, your ECS task execution role must have ECR pull permissions (AmazonECSTaskExecutionRolePolicy) - Deploy your flow to an ECS work pool

from prefect import flow

@flow(log_prints=True)
def my_ecs_flow():
    print("Running on ECS from ECR image")
    return "ok"

if __name__ == "__main__":
    my_ecs_flow.deploy(
        name="my-ecs-ecr-deployment",
        work_pool_name="my-ecs-pool",
        image="<http://123456789012.dkr.ecr.us-east-1.amazonaws.com/my-repo:latest|123456789012.dkr.ecr.us-east-1.amazonaws.com/my-repo:latest>",
        build=False,
        push=False,
        job_variables={
            "cluster": "my-ecs-cluster",
            "launch_type": "FARGATE",
            "cpu": 1024,
            "memory": 2048,
            # Ensure this role has ECR and CloudWatch permissions
            "execution_role_arn": "arn:aws:iam::123456789012:role/ecsTaskExecutionRole",
            # Optionally configure logs:
            # "configure_cloudwatch_logs": True,
            # "cloudwatch_logs_prefix": "prefect",
        },
    )

- Create pool and start worker

prefect work-pool create --type ecs my-ecs-pool
prefect worker start --pool my-ecs-pool --type ecs

Notes: - Docker worker: either log in the host to ECR once, or pass a Prefect

DockerRegistryCredentials

block via

job_variables={"registry_credentials": DockerRegistryCredentials.load("my-ecr-creds")}

. - ECS worker: you usually do not need registry credentials; ECS pulls from ECR using the task execution role. Option B — Start an ECR container from inside a flow (prefect-docker tasks) - Prereqs:

pip install 'prefect[docker]'

and optionally a

DockerRegistryCredentials

block if the host isn’t already logged in to ECR. ``` from prefect import flow from prefect_docker.images import pull_docker_image from prefect_docker.containers import ( create_docker_container, start_docker_container, get_docker_container_logs, remove_docker_container, stop_docker_container ) from prefect_docker import DockerRegistryCredentials, DockerHost @flow def run_ecr_container_in_flow(): # Optional: use Prefect block for ECR auth # ecr_creds = DockerRegistryCredentials.load("my-ecr-creds") docker_host = DockerHost() pull_docker_image(

repository="123456789012.dkr.ecr.us-east-1.amazonaws.com/my-repo", tag="latest", # docker_registry_credentials=ecr_creds, ) container = create_docker_container( image="123456789012.dkr.ecr.us-east-1.amazonaws.com/my-repo:latest", name="my-ecr-container", command=["python", "-c", "print('Hello from ECR image')"], detach=True, docker_host=docker_host, ) start_docker_container(container_id=container.id, docker_host=docker_host) logs = get_docker_container_logs(container_id=container.id, docker_host=docker_host) print(logs) stop_docker_container(container_id=container.id, docker_host=docker_host) remove_docker_container(container_id=container.id, docker_host=docker_host) if name == "__main__": run_ecr_container_in_flow()

Creating a DockerRegistryCredentials block for ECR (if you prefer Prefect-managed auth)

from prefect_docker import DockerRegistryCredentials import boto3, base64 ecr = boto3.client('ecr', region_name='us-east-1') auth = ecr.get_authorization_token() token = auth['authorizationData'][0]['authorizationToken'] username, password = base64.b64decode(token).decode().split(':') registry = auth['authorizationData'][0]['proxyEndpoint'].replace('https://', '') DockerRegistryCredentials( username=username, password=password, registry_url=registry, reauth=True, ).save("my-ecr-creds", overwrite=True) ``` Helpful links - Prefect docs: Prefect Documentation - prefect-docker collection: prefect-docker docs - prefect-aws collection (ECS worker): prefect-aws docs - AWS ECR auth: ECR authentication - ECS task execution role: ECS task execution role If you can share which path you want (Docker worker vs ECS worker, and whether you need to spawn containers from inside a flow), I’ll tailor the exact commands and code to your setup.