Great question. In Prefect 2.20.11 you can programmatically create a deployment with

Deployment.build_from_flow(...).apply()

. Below are minimal examples for both models supported in 2.20.x: - Agent-based (use

infrastructure

+

work_queue_name

) - Worker-based (preferred in 2.20.x; use

work_pool_name

+

job_variables

) I’ve included source links to the exact 2.20.11 implementation for

Deployment.apply

and

Deployment.build_from_flow

at the end. Minimal, agent-based example (Process agent)

from prefect import flow
from prefect.deployments import Deployment
from prefect.infrastructure.process import Process  # agent-based infra block

@flow(name="hello-flow")
def my_flow(name: str = "world"):
    print(f"Hello, {name}!")

deployment = Deployment.build_from_flow(
    flow=my_flow,
    name="hello-deployment",
    infrastructure=Process(),          # agent-based model
    work_queue_name="default",         # queue your agent polls
    parameters={"name": "Marvin"},
    tags=["example"],
)

deployment_id = deployment.apply()     # registers/updates the deployment
print(f"Deployment ID: {deployment_id}")

Minimal, worker-based example (recommended in 2.20.x)

from prefect import flow
from prefect.deployments import Deployment

@flow(name="hello-flow")
def my_flow(name: str = "world"):
    print(f"Hello, {name}!")

deployment = Deployment.build_from_flow(
    flow=my_flow,
    name="hello-deployment",
    work_pool_name="process-pool",     # existing work pool (e.g., type=process, docker, kubernetes)
    job_variables={
        # variables depend on pool type; here are common ones
        "env": {"PREFECT_LOGGING_LEVEL": "DEBUG"}
        # for docker pools you might set: "image": "python:3.11-slim"
    },
    parameters={"name": "Marvin"},
    tags=["example"],
)

deployment_id = deployment.apply()
print(f"Deployment ID: {deployment_id}")

Adding a schedule (2.20.11)

from prefect import flow
from prefect.deployments import Deployment
from prefect.client.schemas.objects import MinimalDeploymentSchedule
from prefect.schedules import CronSchedule

@flow
def my_flow():
    print("Scheduled run")

deployment = Deployment.build_from_flow(
    flow=my_flow,
    name="scheduled-deployment",
    work_pool_name="process-pool",
    schedules=[
        MinimalDeploymentSchedule(
            schedule=CronSchedule(cron="0 9 * * *"),  # 9 AM UTC daily
            active=True,
        ),
    ],
)

deployment_id = deployment.apply()
print(f"Deployment ID: {deployment_id}")

What fields can you set on the deployment? - Common:

name

,

description

,

tags

,

parameters

,

version

- Storage/Code location:

storage

,

path

,

entrypoint

- Scheduling:

schedules

(preferred;

schedule

singular is deprecated) - Agent-based:

infrastructure

(e.g.,

Process

,

DockerContainer

),

work_queue_name

- Worker-based:

work_pool_name

,

job_variables

(nested dict matching your pool’s job template) -

apply(upload=True)

will upload code to configured remote storage if applicable Exact signatures in 2.20.11 (source-verified) -

Deployment.apply(self, upload: bool = False, work_queue_concurrency: int = None) -> UUID

-

Deployment.build_from_flow(cls, flow, name, output=None, skip_upload=False, ignore_file=".prefectignore", apply=False, load_existing=True, schedules=None, **kwargs) -> Deployment

Source links (Prefect 2.20.11) - Deployment.apply - Deployment.build_from_flow - MinimalDeploymentSchedule Notes for CI (e.g., Concourse) - Ensure

PREFECT_API_URL

and (if using Prefect Cloud)

PREFECT_API_KEY

are set in your CI environment

- If using storage/infrastructure that requires extras (e.g., Docker, K8s), install them in CI (e.g.,

pip install 'prefect[docker]'

or

'prefect[kubernetes]'

) Would you like me to tailor the example to agents (work_queue) or workers (work_pool) based on what you’re running today?

Short answer: not with the deployment method you’re using. Because you’re deploying with

Flow.from_source(...).deploy(...)

, you’re using the workers/work-pools model. In that model (Prefect 3.x and the “modern” 2.20.x path), the old

ECSTask

infrastructure block is not used or supported. Instead, you should use an ECS work pool with an ECS worker and pass ECS settings via the deployment’s

job_variables

. If you want to keep using

ECSTask

, you’d have to stick to the legacy 2.x “infrastructure block” deployment path (e.g.,

Deployment.build_from_flow

or the old

prefect deployment build

), both of which are deprecated and removed in 3.x. Since you’re already on the newer deployment style, I’d strongly recommend staying on the worker/work-pool approach. What to do instead (workers/work pool): - Create an ECS work pool and run an ECS worker service - In your

deploy(...)

, set ECS-specific

job_variables

(cluster, task roles, cpu/memory, subnets/security groups, etc.) Example sketch:

deployment_id = await my_flow.deploy(
  name="my-ecs-deployment",
  work_pool_name="my-ecs-pool",
  image="<http://123456789012.dkr.ecr.us-east-1.amazonaws.com/my-image:latest|123456789012.dkr.ecr.us-east-1.amazonaws.com/my-image:latest>",
  job_variables={
    "cluster": "my-ecs-cluster",
    "launch_type": "FARGATE",
    "cpu": 1024,
    "memory": 2048,
    "execution_role_arn": "arn:aws:iam::123456789012:role/ecsTaskExecutionRole",
    "task_role_arn": "arn:aws:iam::123456789012:role/prefectTaskRole",
    "configure_cloudwatch_logs": True,
    "vpc_id": "vpc-abc123",
    "subnets": ["subnet-1", "subnet-2"],
    "security_groups": ["sg-123abc"],
    "env": {"PREFECT_LOGGING_LEVEL": "INFO"}
  }
)

One important thing to clarify: you mentioned you’re on Prefect 2.20.11, but

Flow.from_source(...).deploy(...)

is a 3.x API. It’s possible your Concourse image is running Prefect 3.x while your local is 2.20.11. Could you confirm the version inside Concourse by running:

prefect version

If you share: - The Prefect version reported in Concourse - Your work pool type/name - The exact error/failure you’re seeing - Your current

deploy(...)

call (minus secrets) …I can give you precise ECS

job_variables

and help you resolve the failure quickly.

Yes — with Prefect 2.20.11 and

Deployment.build_from_flow().apply()

, you can use the legacy

ECSTask

infrastructure block from the

prefect-aws

collection. High-level steps: 1) Install the AWS collection in your environment (Concourse image, local CI image, etc.):

pip install prefect-aws

2) Create and save an

ECSTask

block (or load an existing one) 3) Reference that block as the

infrastructure

when you build/apply the deployment 4) Ensure an agent is running to pick up the work queue used by the deployment Example:

from prefect import flow
from prefect.deployments import Deployment
from prefect_aws.ecs import ECSTask
from prefect_aws.credentials import AwsCredentials

@flow
def my_flow():
    print("hello from ECS")

aws_creds = AwsCredentials(
    # Or rely on IAM roles/instance profile in your environment
    aws_access_key_id="...",
    aws_secret_access_key="...",
    region_name="us-east-1",
    # profile_name="my-aws-profile",  # optional alternative
)

ecs = ECSTask(
    aws_credentials=aws_creds,
    cluster="my-ecs-cluster",
    launch_type="FARGATE",
    image="<http://123456789012.dkr.ecr.us-east-1.amazonaws.com/my-image:latest|123456789012.dkr.ecr.us-east-1.amazonaws.com/my-image:latest>",
    cpu=1024,
    memory=2048,
    execution_role_arn="arn:aws:iam::123456789012:role/ecsTaskExecutionRole",
    task_role_arn="arn:aws:iam::123456789012:role/prefectTaskRole",
    configure_cloudwatch_logs=True,
    vpc_id="vpc-abc123",
    subnets=["subnet-1","subnet-2"],
    security_groups=["sg-123abc"],
    env={"PREFECT_LOGGING_LEVEL":"INFO"},
)

# Save once, then you can ECSTask.load("my-ecs-block") in future runs
ecs.save("my-ecs-block", overwrite=True)

Deployment.build_from_flow(
    flow=my_flow,
    name="my-ecs-deployment",
    infrastructure=ecs,              # or ECSTask.load("my-ecs-block")
    work_queue_name="default",       # ensure an agent is serving this queue
    tags=["prod"]
).apply()

Notes and gotchas: - Ensure your execution role can pull your image (ECR permissions) and write logs to CloudWatch if

configure_cloudwatch_logs=True

. - Distinguish

execution_role_arn

(pull image, logs) vs

task_role_arn

(your flow’s AWS calls). - In Concourse, confirm the job has AWS credentials/role available (env vars, profile, or EC2/ECS role). Docs: - Prefect AWS collection (ECSTask): https://prefecthq.github.io/prefect-aws/ecs/ - Deprecation/upgrade context (agents/infrastructure blocks -> workers/work pools): https://docs.prefect.io/latest/guides/upgrade-guide-agents-to-workers/ ECSTask is deprecated and removed in Prefect 3.x. If you plan to upgrade soon, consider moving to an ECS work pool + ECS worker instead. If you share the specific Concourse failure/error logs, I can help pinpoint the exact ECS or IAM setting that needs adjusting.

Yes — in Prefect 2.20.11 you add the entrypoint directly to

Flow.from_source(...)

(or

flow.from_source(...)

). It’s required and points to your flow function. Example (file path entrypoint):

from prefect import flow

flow.from_source(
    source="<https://github.com/org/repo.git>",
    entrypoint="flows/my_flow.py:my_flow",
).deploy(
    name="my-deployment",
    work_pool_name="my-pool",
    build=False,
)

If you prefer a module path, set

entrypoint_type

on `.deploy(...)`:

from prefect.deployments.runner import EntrypointType

flow.from_source(
    source="<https://github.com/org/repo.git>",
    entrypoint="my_package.flows:my_flow",
).deploy(
    name="my-deployment",
    work_pool_name="my-pool",
    entrypoint_type=EntrypointType.MODULE_PATH,
    build=False,
)

Quick note: - File path format:

path/to/file.py:function_name

(default) - Module path format:

package.module:function_name

(use

entrypoint_type=MODULE_PATH

) Docs (Prefect 2.x): - Deployments overview: https://docs-v2.prefect.io/latest/concepts/deployments/ - Work pools: https://docs-v2.prefect.io/latest/concepts/work-pools/ Does this match where you want to add the entrypoint (in your

from_source(...).deploy(...)

call), or were you thinking about setting it via CLI/CI variables?

It’s available in both. Flow.from_source(...) exists in Prefect 2.20.11 (and there’s also the alias flow.from_source(...)), and it continues to be supported in 3.x. - In 2.20.11 the signature is

from_source(source, entrypoint)

and you deploy with

.deploy(...)

. - API reference (2.x): https://docs-v2.prefect.io/api-ref/prefect/flows/#prefect.flows.Flow.from_source - Deployments (2.x): https://docs-v2.prefect.io/latest/concepts/deployments/ If you’re on 2.20.11, you’re using it correctly with

Flow.from_source(...).deploy(...)

.

Absolutely. Here’s a concise Prefect 2.20.11 example that loads your flow from Git and deploys it to an ECS work pool/worker using job_variables (ECSTask): Prereqs - Install the AWS integration where your worker runs:

pip install prefect-aws

- Create a work pool and run a worker (one-time) - Create:

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

- Start:

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

- Ensure your image exists in ECR (or set up build/push in your CI) Example: from_source(...).deploy(...) to ECS

from prefect import flow

# Load your flow from Git and set entrypoint to the flow function
my_flow = flow.from_source(
    source="<https://github.com/your-org/your-repo.git>",
    entrypoint="flows/my_flow.py:my_flow",
)

if __name__ == "__main__":
    my_flow.deploy(
        name="my-ecs-deployment",
        work_pool_name="my-ecs-pool",
        build=False,   # assume your image is already built/pushed
        push=False,
        # ECS-specific configuration goes in job_variables
        job_variables={
            # Cluster and launch type
            "cluster": "my-ecs-cluster",
            "launch_type": "FARGATE",  # or "EC2" or "FARGATE_SPOT"

            # Container image to run
            "image": "<http://123456789012.dkr.ecr.us-east-1.amazonaws.com/my-flow:latest|123456789012.dkr.ecr.us-east-1.amazonaws.com/my-flow:latest>",

            # CPU/Memory for Fargate
            "cpu": 1024,    # 1 vCPU
            "memory": 2048, # 2 GB

            # IAM roles
            "execution_role_arn": "arn:aws:iam::123456789012:role/ecsTaskExecutionRole",
            "task_role_arn": "arn:aws:iam::123456789012:role/ecsTaskRole",

            # Networking (Fargate requires awsvpc)
            "vpc_id": "vpc-0abc1234def567890",
            "network_configuration": {
                "subnets": ["subnet-11111111", "subnet-22222222"],
                "securityGroups": ["sg-0aaa1111bbb2222cc"],
                "assignPublicIp": "ENABLED",  # if you need outbound internet
            },

            # Logging to CloudWatch
            "configure_cloudwatch_logs": True,

            # Environment variables
            "env": {
                "PREFECT_LOGGING_LEVEL": "INFO",
                "MY_APP_ENV": "prod",
            },

            # Optional: reuse family/latest task def
            # "family": "my-task-family",
            # "match_latest_revision_in_family": True,
        },
    )

If your entrypoint is a module path instead of a file path:

from prefect.deployments.runner import EntrypointType

flow.from_source(
    source="<https://github.com/your-org/your-repo.git>",
    entrypoint="my_package.flows:my_flow",
).deploy(
    name="my-ecs-deployment",
    work_pool_name="my-ecs-pool",
    entrypoint_type=EntrypointType.MODULE_PATH,
    build=False,
    job_variables={...}  # same as above
)

Notes and gotchas - Put your container image in

job_variables["image"]

. The ECS worker uses this to run your task. - For CloudWatch logs and ECR pulls, your

execution_role_arn

must allow logs:* and ecr:* actions. - For Fargate, ensure subnets and security groups allow outbound traffic to reach the Prefect API. - If you already have a task definition you prefer to use, set

task_definition_arn

in

job_variables

. Good refs for 2.x - ECS worker guide (2.x source): https://github.com/PrefectHQ/prefect/blob/2.x/docs/integrations/prefect-aws/ecs_guide.md - ECS worker implementation and job_variables schema: https://github.com/PrefectHQ/prefect/blob/2.x/src/integrations/prefect-aws/prefect_aws/workers/ecs_worker.py - Work pools (2.x docs): https://docs-v2.prefect.io/latest/concepts/work-pools/ - Deployments (2.x docs): https://docs-v2.prefect.io/latest/concepts/deployments/ If you share your cluster/VPC setup, I can tailor the

job_variables

block for your exact network and roles.