Short answer: the

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

pattern is supported in Prefect 2.20.11, so the failure in Concourse is almost certainly due to the CI environment. The most common culprits are missing git, missing auth for the repo, missing Prefect API env vars, a non-existent work pool, or attempting to build/push images without Docker privileges in Concourse. A quick checklist to resolve most Concourse CI failures: - Prefect API - Set

PREFECT_API_URL

and

PREFECT_API_KEY

in the task environment. - Confirm network egress to the Prefect API from the worker. - Ensure the work pool exists; either pass

work_pool_name=...

or set

PREFECT_DEFAULT_WORK_POOL_NAME

. - Git and source checkout - If you use

Flow.from_source(...)

with a Git URL, the task container must have

git

installed. - For private repos, pass a token explicitly: -

GitRepository(url="<https://github.com/org/repo.git>", credentials={"access_token": os.environ["GH_TOKEN"]})

- Ensure the

entrypoint="path/to/file.py:flow_fn"

is correct and relative to the repo root. - Docker build/push (only if build=True or push=True) - Concourse tasks are not privileged by default. `docker build`/`push` will fail unless the task is

privileged: true

and you run DinD or have a Docker socket available. - If you can’t enable Docker in Concourse, set

build=False, push=False

and use a prebuilt image in

.deploy(...)

. - Logging - Set

PREFECT_LOGGING_LEVEL=DEBUG

to get a clearer error message. Two reliable patterns in CI A) If Concourse already checks out your repo as a task input (recommended; avoids git in the task) - Use your local flow object and call

.deploy(...)

, skipping `.from_source(...)`:

# deploy.py (executed in the checked-out repo)
import os
from prefect import flow

@flow
def my_flow():
    print("Hello from CI!")

if __name__ == "__main__":
    my_flow.deploy(
        name="my-deployment",
        work_pool_name=os.getenv("PREFECT_WORK_POOL", "my-pool"),
        # Use a pre-built image if Docker is not available in Concourse:
        image=os.getenv("DEPLOY_IMAGE", "myregistry/myimage:latest"),
        build=False,   # Important if you cannot run docker build in Concourse
        push=False
    )

Example Concourse task (no Docker build, repo already checked out):

platform: linux
image_resource:
  type: registry-image
  source: {repository: python, tag: "3.11"}
inputs:
- name: repo
params:
  PREFECT_API_URL: ((prefect_api_url))
  PREFECT_API_KEY: ((prefect_api_key))
  PREFECT_DEFAULT_WORK_POOL_NAME: my-pool
  PREFECT_LOGGING_LEVEL: DEBUG
run:
  path: bash
  args:
  - -exc
  - |
      cd repo
      pip install "prefect==2.20.11"
      python deploy.py

B) If you must use Flow.from_source (remote Git) - Install git and pass a token; keep Docker off unless you’ve enabled privileged tasks:

# deploy_remote.py
import os
from prefect import flow
from prefect.runner.storage import GitRepository

flow_obj = flow.from_source(
    source=GitRepository(
        url="<https://github.com/org/repo.git>",
        # Prefer a token over SSH in CI
        credentials={"access_token": os.environ["GH_TOKEN"]},
        # optionally: reference="main"
    ),
    entrypoint="flows.py:my_flow",
)

flow_obj.deploy(
    name="my-deployment",
    work_pool_name=os.getenv("PREFECT_WORK_POOL", "my-pool"),
    image=os.getenv("DEPLOY_IMAGE", "myregistry/myimage:latest"),
    build=False,
    push=False,
)

Concourse task: ``` platform: linux image_resource: type: registry-image source: {repository: python, tag: "3.11"} params: PREFECT_API_URL: ((prefect_api_url)) PREFECT_API_KEY: ((prefect_api_key)) PREFECT_DEFAULT_WORK_POOL_NAME: my-pool GH_TOKEN: ((github_token)) PREFECT_LOGGING_LEVEL: DEBUG run: path: bash args: - -exc - | pip install "prefect==2.20.11" # install git if base image doesn't include it if ! command -v git; then apt-get update && apt-get install -y git fi

python deploy_remote.py ``` If you need to build/push images in Concourse - You must run the task as

privileged: true

and run Docker-in-Docker (or mount a Docker socket) and then set

build=True, push=True

. Many teams instead prebuild the image in a separate pipeline step and pass that image reference to

.deploy(...)

with

build=False, push=False

. Common errors and fixes - “Could not find work pool ‘X’”: Create the work pool first in the Prefect UI or via API/CLI, or set the correct name. - “fatal: Authentication failed” or “Could not read from remote repository”: Install

git

in the task image and pass a token via `credentials={"access_token": ...}`; avoid SSH unless you manage keys/agent in CI. - “docker: command not found” or permission denied: Either enable privileged tasks and Docker, or set

build=False, push=False

and use a prebuilt image. - Silent failures or unclear logs: Add

PREFECT_LOGGING_LEVEL=DEBUG

. If you can share the exact error message Concourse returns and (optionally) your task config snippet, I can pinpoint the failing step and give a focused fix.

Got it — if you’re using an

S3Bucket

block with

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

in Prefect 2.20.11, the CI failures in Concourse are usually one (or more) of the following: What to verify in Concourse - Install the AWS integration - You must install

prefect-aws

in the Concourse task image or step. Without it,

S3Bucket.load(...)

will fail. - Example:

pip install "prefect==2.20.11" "prefect-aws>=0.3"

- Make sure the

S3Bucket

block can authenticate - If your block embeds credentials via an

AwsCredentials

block, that’s fine — just ensure

prefect-aws

is installed so it can be loaded. - If not, provide AWS credentials via environment variables in the task: -

AWS_ACCESS_KEY_ID

,

AWS_SECRET_ACCESS_KEY

, optional

AWS_SESSION_TOKEN

, and

AWS_DEFAULT_REGION

- Entry point and bucket layout - Your

entrypoint

must be correct and relative to the block’s configured folder/prefix (e.g., if

bucket_folder="code/"

, your entrypoint should be something like

code/my_flow.py:my_flow

). - Ensure the entire module tree your flow imports exists under that prefix in S3. - Prefect API config - Set

PREFECT_API_URL

and

PREFECT_API_KEY

for Prefect Cloud (or the server URL/key for self-hosted). - The work pool you pass in

work_pool_name

must already exist (or set

PREFECT_DEFAULT_WORK_POOL_NAME

). - Docker build/push - If your Concourse task cannot run Docker (normal case), use a prebuilt image and disable build/push: -

image="myregistry/myimage:tag", build=False, push=False

- If you do need to build/push in CI, the task must be privileged and have Docker available. Minimal working example with S3Bucket

# deploy_s3.py
import os
from prefect import flow
from prefect_aws.s3 import S3Bucket

@flow
def my_flow():
    print("Hello from S3 source")

if __name__ == "__main__":
    # Load a configured block (requires prefect-aws installed)
    s3 = S3Bucket.load("my-code-bucket")  # bucket_name and bucket_folder set in the block

    # entrypoint is relative to the block's bucket_folder
    f = flow.from_source(
        source=s3,
        entrypoint="flows/my_flow.py:my_flow"
    )

    f.deploy(
        name="my-deployment",
        work_pool_name=os.getenv("PREFECT_WORK_POOL", "my-pool"),
        # Prefer a prebuilt image in CI to avoid Docker build/push
        image=os.getenv("DEPLOY_IMAGE", "myregistry/myimage:latest"),
        build=False,
        push=False,
    )

Example Concourse task snippet

platform: linux
image_resource:
  type: registry-image
  source: {repository: python, tag: "3.11"}
params:
  PREFECT_API_URL: ((prefect_api_url))
  PREFECT_API_KEY: ((prefect_api_key))
  PREFECT_DEFAULT_WORK_POOL_NAME: my-pool
  # If your block doesn't embed creds, provide them here
  AWS_ACCESS_KEY_ID: ((aws_access_key_id))
  AWS_SECRET_ACCESS_KEY: ((aws_secret_access_key))
  AWS_DEFAULT_REGION: us-east-1
  PREFECT_LOGGING_LEVEL: DEBUG
run:
  path: bash
  args:
  - -exc
  - |
      pip install "prefect==2.20.11" "prefect-aws>=0.3"
      python deploy_s3.py

Common S3-related pitfalls - ModuleNotFoundError at deploy-time/run-time: ensure all Python modules your flow imports are present under the same S3 prefix; the worker pulls code from S3 at run-time. - Wrong entrypoint path: must be relative to the

bucket_folder

set on the block. - Missing prefect-aws at runtime:

S3Bucket.load(...)

will fail without the integration installed in the CI image. - Missing AWS credentials: set env vars or embed credentials in the

S3Bucket

via an

AwsCredentials

block. If you can paste the exact error Concourse shows (stack trace or message) and how you’re loading the

S3Bucket

(block name,

bucket_folder

, and your

entrypoint

), I can pinpoint the issue quickly. Docs for reference: - AWS S3 block (prefect-aws): S3Bucket block docs - S3 utilities: prefect-aws S3 docs

Thanks — when a Concourse task “hangs” at

.deploy(...)

in Prefect 2.20.11 with S3Bucket storage, it’s almost always one of these: 1) Docker build/push running in the task - By default

.deploy()

does

build=True, push=True

. If your Concourse task isn’t privileged or can’t reach Docker/registries, it can appear to hang while trying to build/pull base images or push. - Quick fix: disable build/push and use a prebuilt image.

f.deploy(
    name="my-deployment",
    work_pool_name="my-pool",
    image="myregistry/myimage:tag",
    build=False,
    push=False,
  )

2) S3 download/materialization hanging on AWS credential discovery - When using an

S3Bucket

block, Prefect pulls code from S3 as part of the deploy flow. If AWS creds are not explicitly provided, boto3 will try multiple providers including the EC2 Instance Metadata Service (IMDS). In a CI container, IMDS lookups can hang for a long time. - Fixes: - Ensure the block embeds credentials (preferred), or set env vars in the task:

AWS_ACCESS_KEY_ID

,

AWS_SECRET_ACCESS_KEY

,

AWS_DEFAULT_REGION

(and

AWS_SESSION_TOKEN

if needed). - Set

AWS_EC2_METADATA_DISABLED=true

to prevent IMDS lookups in CI. 3) Prefect API connectivity blocked -

.deploy()

talks to your Prefect API. If egress is blocked or env vars are missing, it can stall on network retries. - Verify

PREFECT_API_URL

and

PREFECT_API_KEY

are set and reachable from the task (e.g.,

curl $PREFECT_API_URL/health

). A quick isolate-and-fix plan - Step 1: Disable image build/push to see if the hang disappears.

f.deploy(name="my-deployment", work_pool_name="my-pool",
           image="python:3.11-slim", build=False, push=False)

If this completes quickly, the hang was Docker/registry related. Keep build/push disabled in Concourse or switch to a privileged task with working Docker. - Step 2: Confirm S3 access won’t block - In your Concourse task params:

AWS_ACCESS_KEY_ID: ((aws_access_key_id))
    AWS_SECRET_ACCESS_KEY: ((aws_secret_access_key))
    AWS_DEFAULT_REGION: us-east-1
    AWS_EC2_METADATA_DISABLED: "true"

- Also verify your

entrypoint

is correct and relative to the S3 block’s

bucket_folder

. - Step 3: Turn on debug logs to pinpoint the step - Add

PREFECT_LOGGING_LEVEL=DEBUG

- Optional: enable boto logs to confirm IMDS/S3 behavior

import logging, boto3
    boto3.set_stream_logger("boto3", logging.DEBUG)
    boto3.set_stream_logger("botocore", logging.DEBUG)

Minimal Concourse task example

platform: linux
image_resource:
  type: registry-image
  source: {repository: python, tag: "3.11"}
params:
  PREFECT_API_URL: ((prefect_api_url))
  PREFECT_API_KEY: ((prefect_api_key))
  PREFECT_DEFAULT_WORK_POOL_NAME: my-pool

  # S3 auth (if not embedded in the block)
  AWS_ACCESS_KEY_ID: ((aws_access_key_id))
  AWS_SECRET_ACCESS_KEY: ((aws_secret_access_key))
  AWS_DEFAULT_REGION: us-east-1
  AWS_EC2_METADATA_DISABLED: "true"

  PREFECT_LOGGING_LEVEL: DEBUG
run:
  path: bash
  args:
  - -exc
  - |
      pip install "prefect==2.20.11" "prefect-aws>=0.3"
      python deploy_s3.py

deploy_s3.py sketch

import os, logging
from prefect import flow
from prefect_aws.s3 import S3Bucket

logging.basicConfig(level=os.getenv("PREFECT_LOGGING_LEVEL","INFO"))

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

if __name__ == "__main__":
    s3 = S3Bucket.load("my-code-bucket")
    f = flow.from_source(source=s3, entrypoint="flows/my_flow.py:my_flow")
    f.deploy(
        name="my-deployment",
        work_pool_name=os.getenv("PREFECT_WORK_POOL", "my-pool"),
        image=os.getenv("DEPLOY_IMAGE", "python:3.11-slim"),
        build=False,
        push=False,
    )

If it still hangs after disabling build/push and setting AWS env vars, please paste: - The last 30–50 lines of Concourse logs around

.deploy(...)

- Whether your S3Bucket block embeds credentials or relies on env vars

- Your

entrypoint

and the block’s

bucket_folder

- Whether you intend to build/push images inside Concourse Useful docs: - S3Bucket block docs: prefect-aws S3Bucket - S3 integrations: prefect-aws S3

Short answer: Yes. In Prefect 2.20.11, you can use

Deployment.build_from_flow()

and it still works. However, it’s deprecated in 2.x and removed in 3.x. Expect deprecation warnings, and plan to migrate to

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

and

prefect deploy

when you move to Prefect 3.x. Minimal example for 2.20.11:

from prefect import flow
from prefect.deployments import Deployment

@flow
def my_flow():
    ...

Deployment.build_from_flow(flow=my_flow, name="my-deployment", apply=True)

If you’re already using

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

, that’s the 3.x pattern and won’t work on 2.20.11. For 2.20.11, stick with

Deployment.build_from_flow()

or the 2.x deployment CLI until you upgrade. References: - Upgrade/migration guidance: Upgrade guide (agents → workers) - Background on 2.x deployments: Deployments (2.x concepts) If you want, I can help translate your current Concourse CI deployment step to the 2.20.11-safe approach or outline the migration path to 3.x.

That error is a generic wrapper in 2.20.11 that hides the real cause. It means the worker failed to import or locate the flow for the deployment’s entrypoint. The actual exception only appears in the worker logs (not in the UI). Quick triage - Start a worker with debug logs and watch its stderr:

prefect worker start --work-pool YOUR_POOL --log-level DEBUG

- Confirm the entrypoint loads locally:

python - <<'PY'
from prefect.flows import load_flow_from_entrypoint
load_flow_from_entrypoint("relative/path/to/flow.py:flow_function_name")
print("OK")
PY

Common causes and fixes (2.20.11) 1) Entrypoint path mismatch - The entrypoint must be like

relative/path/to/flow.py:my_flow

. -

Deployment.build_from_flow()

will auto-calculate it; print it to confirm:

dep = await Deployment.build_from_flow(flow=my_flow, name="prod")
print(dep.entrypoint)  # should be relative path + ":my_flow"

2) Using skip_upload=True without shared code - If your worker runs on a different machine (very likely with Concourse), do NOT set

skip_upload=True

unless the code is already on the worker (e.g., baked into a Docker image). - Fix: use a filesystem/storage block and set

skip_upload=False

so workers can fetch code. Example with S3 storage block

from prefect import flow
from prefect.deployments import Deployment
from prefect.filesystems import S3
import asyncio

@flow
def my_flow():
    return "ok"

async def main():
    s3 = await S3.load("my-s3-block")  # points at your bucket/prefix
    await Deployment.build_from_flow(
        flow=my_flow,
        name="prod",
        storage=s3,
        skip_upload=False,  # upload code so remote workers can fetch it
        apply=True
    )

asyncio.run(main())

3) .prefectignore excludes your flow files - If

*.py

or broad patterns are present, your flow may be excluded from upload. - Ensure your flow files are included, e.g.:

__pycache__/
*.pyc
venv/
.git/
# allow flows
!flows/*.py
!src/**/*.py

4) Missing dependencies on the worker - If the flow imports libs not present in the worker environment, import will fail. - Fix: install dependencies in the worker environment or use a Docker work pool with an image that includes your requirements. 5) Build dir vs runtime dir - The entrypoint is relative to the project root used at build time; workers fetch code into a working directory. - Ensure you build from the repo root (so relative paths resolve consistently), and avoid overriding entrypoint to absolute paths. 6) Concourse CI gotchas - CI is ephemeral; don’t rely on

skip_upload=True

. - Build/apply from the repo root to produce a clean, relative

entrypoint

. - Ensure the storage block (e.g., S3/GCS/Azure) is configured and reachable from the worker. - If you’re using Docker work pools, either bake your code into the image or still use storage so workers can pull the latest code. If you prefer CLI on 2.20.11

prefect deployment build ./flows/my_flow.py:my_flow \
  --name prod \
  --storage-block s3/my-s3-block \
  --apply

What I need to help you pinpoint it - The exact worker log error (from

prefect worker start --work-pool ... --log-level DEBUG

) - How you’re building in Concourse (snippet of your build/apply step) - Your deployment code/CLI and whether

skip_upload

is set - Your

.prefectignore

- The printed

deployment.entrypoint

after

build_from_flow

References - 2.x deployments background: Deployments (2.x) - Upgrade guidance (context on deprecations): Upgrade guide

Short answer: in Prefect 2.20.11 you programmatically create a deployment by building a Deployment object from your flow, then calling

.apply()

to register it with the API. Minimal complete example

from prefect import flow
from prefect.deployments import Deployment

@flow
def my_flow(name: str = "World"):
    print(f"Hello {name}!")

# Build a deployment for the flow
deployment = Deployment.build_from_flow(
    flow=my_flow,
    name="hello-deployment",
    # Optional (2.20.11 supports both of these):
    work_pool_name="my-pool",      # the work pool your worker uses
    work_queue_name="default",     # the queue within that pool (defaults to "default")
    # parameters={"name": "Marvin"},  # default run parameters if desired
)

# Register the deployment with the Prefect API
deployment_id = deployment.apply()        # or deployment.apply(upload=True) if using remote storage
print(f"Deployment created: {deployment_id}")

Common options you may want - Default parameters for your flow runs

deployment = Deployment.build_from_flow(
    flow=my_flow,
    name="with-params",
    parameters={"name": "Prefect"},
)
deployment.apply()

- Use remote storage and upload code (recommended for CI like Concourse)

from prefect.filesystems import S3

deployment = Deployment.build_from_flow(
    flow=my_flow,
    name="s3-storage",
    storage=S3.load("my-bucket-block"),  # preconfigured S3 block
    path="flows/",                       # subdirectory within the storage
)
deployment.apply(upload=True)            # pushes code to S3

- Choose infrastructure (e.g., Docker)

from prefect.infrastructure import DockerContainer

deployment = Deployment.build_from_flow(
    flow=my_flow,
    name="docker",
    infrastructure=DockerContainer(
        image="python:3.11",
        auto_remove=True,
    ),
    # job_variables can override runtime settings like env, memory, etc.
    job_variables={
        "env.PREFECT_LOG_LEVEL": "DEBUG",
    },
)
deployment.apply()

- Add a schedule (2.20.11 prefers the plural

schedules

list)

from prefect.client.schemas.schedules import CronSchedule
from prefect.client.schemas.objects import MinimalDeploymentSchedule

deployment = Deployment.build_from_flow(
    flow=my_flow,
    name="scheduled",
    schedules=[
        MinimalDeploymentSchedule(
            schedule=CronSchedule(cron="0 9 * * MON-FRI"),
            active=True,
        )
    ],
)
deployment.apply()

Notes specific to 2.20.11 -

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

is the standard 2.x programmatic path. Both methods are sync-compatible, so you can call them without

await

. -

work_pool_name

and

work_queue_name

are supported; if you’re using workers (recommended in 2.20.11), set the pool/queue your worker is polling. -

job_variables

is the preferred name for per-run infrastructure overrides (older

infra_overrides

still works in 2.x YAML). - If running from ephemeral CI (e.g., Concourse), use remote

storage

and call

apply(upload=True)

so your code is available to workers at run time. - The

Deployment

class is deprecated in favor of

flow.deploy()

(and in Prefect 3.x the

Deployment

class is removed). Since you asked about 2.20.11, the examples above reflect that version. References - API reference: Deployment - Concepts: deployments, storage, infrastructure - Guide: move from agents to workers