Thanks for the details — the error makes sense given the current deploy script. What’s happening - Your deployment is being created without any source/storage info, so at runtime the worker doesn’t know where to find your flow code. - That’s why you see “Downloading flow code from storage at None” and then a FileNotFoundError like “/workspace/None”. - In Prefect 3.x, a deployment must include instructions for how the worker will get the code at run time (aka “pull steps”). You provide these by calling

Flow.from_source(...)

before

.deploy(...)

. Two ways to fix it A) Use Git (recommended for reproducibility) - Tell Prefect where to pull the code from (your repo) and which file:function defines the flow.

import os
import json
import sys
from prefect.flows import Flow
from prefect.docker.docker_image import DockerImage

def main():
    if len(sys.argv) != 2:
        print("Usage: python deploy.py <config_path>")
        sys.exit(1)

    config_path = sys.argv[1]
    with open(config_path, "r") as f:
        config = json.load(f)

    tags = config.get("tags", [])
    commit = os.getenv("GITHUB_SHA")
    if commit:
        tags.append(f"commit:{commit[:8]}")

    # Point to your Git repo and entrypoint (e.g., "flows/my_flow.py:my_flow")
    flow = Flow.from_source(
        source=config["git_url"],           # e.g., "<https://github.com/owner/repo.git>"
        entrypoint=config["entrypoint"],    # e.g., "flows/my_flow.py:my_flow"
    )

    docker_image = DockerImage(
        name="cyclopsai/template",
        dockerfile=config["docker_image"],  # path to your Dockerfile (or "auto")
    )

    flow.deploy(
        name=config["name"],
        work_pool_name=config["work_pool_name"],
        description=config.get("description", ""),
        tags=tags,
        push=True,
        image=docker_image,
        parameters=config.get("parameters", {}),
    )

if __name__ == "__main__":
    main()

Result: Prefect will add a

git_clone

pull step to each flow run so the worker can fetch your code. B) Bake the code into the image (no Git clone at runtime) - This works great, but you still need to tell Prefect where the code will be inside the container. - Use

Flow.from_source(".", entrypoint=...)

so Prefect records the relative path and sets the working directory at runtime. - Ensure your Dockerfile copies the project into the same path Prefect expects (by default

/opt/prefect/flows

if you use the “auto” builder). Example:

import os
import json
import sys
from prefect.flows import Flow
from prefect.docker.docker_image import DockerImage

def main():
    if len(sys.argv) != 2:
        print("Usage: python deploy.py <config_path>")
        sys.exit(1)

    config_path = sys.argv[1]
    with open(config_path, "r") as f:
        config = json.load(f)

    tags = config.get("tags", [])
    commit = os.getenv("GITHUB_SHA")
    if commit:
        tags.append(f"commit:{commit[:8]}")

    # Record current directory as the source and the flow entrypoint
    flow = Flow.from_source(
        source=".",                         # local project root
        entrypoint=config["entrypoint"],    # e.g., "flows/my_flow.py:my_flow"
    )

    docker_image = DockerImage(
        name="cyclopsai/template",
        dockerfile=config["docker_image"],  # path to your Dockerfile OR "auto"
    )

    flow.deploy(
        name=config["name"],
        work_pool_name=config["work_pool_name"],
        description=config.get("description", ""),
        tags=tags,
        push=True,
        image=docker_image,
        parameters=config.get("parameters", {}),
        # optionally, ensure the worker's working dir matches your Dockerfile
        # job_variables={"working_dir": "/opt/prefect/flows"},
    )

if __name__ == "__main__":
    main()

If you use a custom Dockerfile, make sure it copies your code into

/opt/prefect/flows

(or set

job_variables["working_dir"]

to whatever path you use): ``` FROM prefecthq/prefect:3-python3.12

WORKDIR /opt/prefect/flows COPY . /opt/prefect/flows RUN pip install -r requirements.txt ``` Why your current script fails -

load_flow_from_entrypoint(...)

creates a Flow, but without

from_source(...)

Prefect does not have pull steps or a path to set at runtime. The deployment ends up with

storage=None

and

path=None

, leading to

/workspace/None

. Docs for reference - Store your flow code: Store flow code - Docker deployments: Docker deployments - Deployment concepts (pull steps, entrypoint, path): Deployment concepts If you can share: - Your

config["entrypoint"]

value - Whether you’re using a custom Dockerfile and its relevant lines (COPY/WORKDIR) - Which work pool/worker type you’re using …I can tailor the exact

from_source(...)

and Dockerfile/`job_variables` you should use.