Short answer: yes. In Prefect 3.x you can use a “flow factory” as long as the factory produces concrete, importable flow objects at module import time so they can be referenced by an entrypoint like file.py:attr or package.module:attr. Entrypoints must resolve to a Flow or a plain function that Prefect can convert to a Flow; they cannot be a call expression (e.g., package.module:make_flow("x")). Two reliable patterns 1) Materialize flows at import time (classic factory) - Use a factory to generate actual flow objects and bind them to module-level names. Then each can be referenced in a deployment entrypoint.

# my_project/flows/factory.py
from prefect import flow

def make_flow(name: str, multiplier: int):
    @flow(name=name)
    def dynamic_flow(x: int = 1):
        return x * multiplier
    return dynamic_flow

# Materialize variants at import time so they are discoverable by attribute
flow_alpha = make_flow("alpha-flow", multiplier=2)
flow_beta = make_flow("beta-flow", multiplier=3)

Now each is discoverable with an entrypoint: - file path form:

my_project/flows/factory.py:flow_alpha

- module path form (if your package is installed/importable):

my_project.flows.factory:flow_alpha

Deploy examples: - Via prefect.yaml

deployments:
  - name: alpha-deploy
    entrypoint: my_project/flows/factory.py:flow_alpha
    work_pool_name: docker-pool
    image: <http://ghcr.io/org/prefect-flows:latest|ghcr.io/org/prefect-flows:latest>

  - name: beta-deploy
    entrypoint: my_project/flows/factory.py:flow_beta
    work_pool_name: docker-pool
    image: <http://ghcr.io/org/prefect-flows:latest|ghcr.io/org/prefect-flows:latest>

- Or via Python (build/push image automatically):

from prefect.docker import DockerImage

# import flow_alpha and flow_beta from your module
from my_project.flows.factory import flow_alpha, flow_beta

flow_alpha.deploy(
    name="alpha-deploy",
    work_pool_name="docker-pool",
    image=DockerImage(name="<http://ghcr.io/org/prefect-flows:latest|ghcr.io/org/prefect-flows:latest>", dockerfile="Dockerfile"),
)

flow_beta.deploy(
    name="beta-deploy",
    work_pool_name="docker-pool",
    image=DockerImage(name="<http://ghcr.io/org/prefect-flows:latest|ghcr.io/org/prefect-flows:latest>", dockerfile="Dockerfile"),
)

2) Registry + single router flow (scale with parameters) - Keep one entrypoint and select “which flow” via parameters. Then create multiple deployments that point at the same entrypoint but with different parameter sets.

# my_project/flows/registry.py
from prefect import flow

@flow
def worker_a(x: int = 1): return x * 2

@flow
def worker_b(x: int = 1): return x * 3

REGISTRY = {"alpha": worker_a, "beta": worker_b}

@flow
def router(which: str, x: int = 1):
    return REGISTRY<x|which>

Deploy two variants with the same entrypoint but different parameters:

deployments:
  - name: alpha-deploy
    entrypoint: my_project/flows/registry.py:router
    parameters:
      which: "alpha"
      x: 5
    work_pool_name: docker-pool
    image: <http://ghcr.io/org/prefect-flows:latest|ghcr.io/org/prefect-flows:latest>

  - name: beta-deploy
    entrypoint: my_project/flows/registry.py:router
    parameters:
      which: "beta"
      x: 10
    work_pool_name: docker-pool
    image: <http://ghcr.io/org/prefect-flows:latest|ghcr.io/org/prefect-flows:latest>

Key gotchas and notes - Entrypoint must be a static attribute path resolving to either: - a Flow object (e.g.,

my_module:flow_alpha

), or - a plain function Prefect will convert to a Flow. It cannot be an expression like

my_module:make_flow('alpha')

. - For Docker deployments, make sure: - Your code is copied into the image at build time and importable at runtime. - If you use a module path entrypoint, install your package in the image (e.g., pip install .) or set PYTHONPATH accordingly. - Set the deployment’s image to your built image (via Python

image=...

or in prefect.yaml). - File path vs module path: - CLI help shows file path format (

./path/to/file.py:flow_func

). - Module paths generally work when using the Python API or when your package is installed and importable in the container. If in doubt for CLI, prefer file path form.

- If your flows live in a remote repo, use Flow.from_source(...).deploy(...) with an entrypoint string:

from prefect import Flow

Flow.from_source(
    source="<https://github.com/org/repo>",
    entrypoint="src/my_project/flows/factory.py:flow_alpha",
).deploy(name="alpha-deploy", work_pool_name="docker-pool")

Helpful docs - Deployments overview: Docs: Deployments - Create deployments (Python and CLI): How-to: Create deployments - Project/prefect.yaml: How-to: prefect.yaml - Docker guidance: How-to: Docker deployments Does one of these patterns fit what you mean by “flow factory”? If you share how you’re generating flows (e.g., a loop that creates dozens of variants), I can suggest the cleanest entrypoint and deployment config for your Docker image.

You’ve hit a real gotcha with “factory-made” flows: Prefect derives the entrypoint from the function’s defining module and function name, not from the variable you assign it to later. Because your

dynamic_flow

is defined inside

factory.make_flow

, it ends up with: - module = "factory" - qualname = "make_flow.<locals>.dynamic_flow" Then when Prefect tries to compute an entrypoint for deployment, it cannot import

factory:dynamic_flow

(it’s a local, not a top-level symbol) and it ignores the

impl.flow_alpha

binding. You have three solid ways to fix this: Option A: Rebind the flow’s underlying function metadata after creation Do this in the module that will be used as the entrypoint (impl.py). The Flow object has a

.fn

attribute; set its module/name to match the symbol you’re exporting.

# factory.py
from prefect import flow

def make_flow(name: str, multiplier: int):
    @flow(name=name)
    def dynamic_flow(x: int = 1):
        return x * multiplier
    return dynamic_flow

# impl.py
from .factory import make_flow

flow_alpha = make_flow("alpha-flow", multiplier=2)
flow_alpha.fn.__name__ = "flow_alpha"
flow_alpha.fn.__qualname__ = "flow_alpha"
flow_alpha.fn.__module__ = __name__

flow_beta = make_flow("beta-flow", multiplier=3)
flow_beta.fn.__name__ = "flow_beta"
flow_beta.fn.__qualname__ = "flow_beta"
flow_beta.fn.__module__ = __name__

# Now these entrypoints work:
# - ./impl.py:flow_alpha
# - ./impl.py:flow_beta

Tip: You can validate your entrypoint resolves with:

from prefect.flows import load_flow_from_entrypoint
load_flow_from_entrypoint("path/to/impl.py:flow_alpha")

Option B: Have the factory set the caller’s module and the export name If you want to avoid manual rebinding in every module that uses the factory, make the factory do it for you.

# factory.py
import inspect
from prefect import flow

def make_flow(export_name: str, display_name: str, multiplier: int):
    @flow(name=display_name)
    def _inner(x: int = 1):
        return x * multiplier

    # Rebind to the caller's module under the desired export name
    caller_module = inspect.getmodule(inspect.stack()[1][0]).__name__
    _inner.fn.__name__ = export_name
    _inner.fn.__qualname__ = export_name
    _inner.fn.__module__ = caller_module
    return _inner

# impl.py
from .factory import make_flow

flow_alpha = make_flow("flow_alpha", "alpha-flow", 2)
flow_beta  = make_flow("flow_beta",  "beta-flow",  3)

Option C: Skip factory-materialized flows; use parameters and multiple deployments Keep one top-level flow and create multiple deployments with different parameters (simplest and most robust, especially for Docker).

# flows.py
from prefect import flow

@flow
def multiplier_flow(multiplier: int, x: int = 1):
    return x * multiplier

prefect.yaml:

deployments:
  - name: alpha-deploy
    entrypoint: src/my_project/flows.py:multiplier_flow
    parameters:
      multiplier: 2
    work_pool_name: docker-pool
    image: <http://ghcr.io/org/prefect-flows:latest|ghcr.io/org/prefect-flows:latest>

  - name: beta-deploy
    entrypoint: src/my_project/flows.py:multiplier_flow
    parameters:
      multiplier: 3
    work_pool_name: docker-pool
    image: <http://ghcr.io/org/prefect-flows:latest|ghcr.io/org/prefect-flows:latest>

Notes - Entrypoints must resolve to a top-level attribute that can be imported. Local/nested function names like

make_flow.<locals>.dynamic_flow

are not importable and will fail. - If you’re deploying programmatically with

flow.to_deployment(...)

, the entrypoint is inferred via introspection; that’s why the metadata rebinding above is necessary for factory-made flows. - If you use the CLI with an explicit entrypoint (recommended), Prefect will use what you specify: -

prefect deploy ./src/my_project/impl.py:flow_alpha

- CLI help: Docs: CLI flow and How-to: Create deployments

If you share how you’re deploying (CLI with prefect.yaml vs Python API) and how you’re building the Docker image, I can tailor the exact steps and config to your setup.

You’re running into Prefect’s parameter serialization rules: deployment parameters must be JSON-serializable. Python modules/classes/functions aren’t. The pattern for “one generic flow, many deployments” is to encode the implementation you want as data (a string or enum), then resolve it inside the flow at runtime. Two solid patterns that work well with option C: Pattern 1: Pass a qualified import path (string), import at runtime - Pass something like "my_pkg.impls.alpha:AlphaImpl" or "my_pkg.impls.alpha:run" as a string parameter. - In the flow, resolve it with importlib or Prefect’s helper and then call it.

# flows.py
from prefect import flow
from prefect.importtools import import_object  # resolves "pkg.module:attr"

@flow
def generic_flow(impl_path: str, params: dict | None = None):
    obj = import_object(impl_path)
    params = params or {}

    # If it's a class, instantiate then call .run(); if it's a function, call it
    if isinstance(obj, type):
        instance = obj(**params)
        return instance.run() if hasattr(instance, "run") else instance(**params)
    else:
        return obj(**params)

prefect.yaml with multiple deployments:

deployments:
  - name: alpha-deploy
    entrypoint: src/my_project/flows.py:generic_flow
    parameters:
      impl_path: "my_pkg.impls.alpha:AlphaImpl"
      params: {"x": 2}
    work_pool_name: docker-pool
    image: <http://ghcr.io/org/prefect-flows:latest|ghcr.io/org/prefect-flows:latest>

  - name: beta-deploy
    entrypoint: src/my_project/flows.py:generic_flow
    parameters:
      impl_path: "my_pkg.impls.beta:run"
      params: {"x": 3}
    work_pool_name: docker-pool
    image: <http://ghcr.io/org/prefect-flows:latest|ghcr.io/org/prefect-flows:latest>

Notes - The container image must include your code so these import paths resolve at runtime. - This avoids passing non-serializable Python objects through deployment parameters. - It also sidesteps your polymorphism/qualname concern: you choose the exact subclass or function via the string path. Pattern 2: Registry + key (works nicely with polymorphism) - Subclasses or implementations register themselves into a module-level REGISTRY at import time. - You pass a small serializable key (string/enum) and any JSON params; the flow looks up the implementation by key.

# registry.py
REGISTRY: dict[str, object] = {}

def register(name: str):
    def _wrap(obj):
        REGISTRY[name] = obj
        return obj
    return _wrap

# impl_alpha.py
from .registry import register

@register("alpha")
class AlphaImpl:
    def __init__(self, x: int = 1): self.x = x
    def run(self): return self.x * 2

# impl_beta.py
from .registry import register

@register("beta")
def run(x: int = 1): return x * 3

# flows.py
from prefect import flow
from my_project.registry import REGISTRY
import my_project.impl_alpha  # ensure modules are imported so registration runs
import my_project.impl_beta

@flow
def generic_flow(which: str, params: dict | None = None):
    impl = REGISTRY[which]
    params = params or {}
    if isinstance(impl, type):
        inst = impl(**params)
        return inst.run() if hasattr(inst, "run") else inst(**params)
    else:
        return impl(**params)

prefect.yaml:

deployments:
  - name: alpha-deploy
    entrypoint: src/my_project/flows.py:generic_flow
    parameters:
      which: "alpha"
      params: {"x": 5}
    work_pool_name: docker-pool
    image: <http://ghcr.io/org/prefect-flows:latest|ghcr.io/org/prefect-flows:latest>

  - name: beta-deploy
    entrypoint: src/my_project/flows.py:generic_flow
    parameters:
      which: "beta"
      params: {"x": 10}
    work_pool_name: docker-pool
    image: <http://ghcr.io/org/prefect-flows:latest|ghcr.io/org/prefect-flows:latest>

Why this solves your polymorphism caveat - You noted that a function’s module/qualname reflect where it’s defined (e.g., base class), not necessarily the subclass that’s invoking it. Both patterns above avoid relying on introspecting bound methods. - Pattern 1 explicitly imports the target class/function by string path you choose at deployment time.

- Pattern 2 uses explicit registration in the subclass module; you reference the subclass by a stable key. Extras and tips - If you don’t want to store the impl_path in the deployment, you can read it from an environment variable or a Prefect Variable inside the flow. - You can strongly type params by parsing it into a Pydantic model inside the flow. - Validate your impl_path early with:

from prefect.importtools import import_object
import_object("my_pkg.impls.alpha:AlphaImpl")  # raises if invalid

Relevant docs - Deployments overview: Docs: Deployments - Create deployments (Python and CLI): How-to: Create deployments - Docker guidance: How-to: Docker deployments If you share an example of the module/class/function you’re trying to pass, I can sketch the exact impl_path/registry wiring and a matching prefect.yaml for your image.

Short answer: with Flow.to_deployment() and Flow.deploy(), the entrypoint is inferred from the flow’s underlying function (module + name). There isn’t a parameter on those methods to override the entrypoint string. Ways to set an explicit entrypoint from Python (works with Docker): - Use the top-level deploy() function with an entrypoint string This lets you specify the entrypoint directly and still target a Docker work pool. You set the image via work pool defaults or job_variables.

from prefect import deploy

deploy(
  entrypoint="src/my_project/impl.py:flow_alpha",
  names=["alpha-deploy"],
  work_pool_name="docker-pool",
  # for Docker, either set image on the work pool or pass it here
  job_variables={"image": "<http://ghcr.io/acme/prefect-flows:latest|ghcr.io/acme/prefect-flows:latest>"},
  # parameters=... etc
)

Notes: - This path is explicit and avoids module/qualname inference. - If you need to build/push the image, do it yourself (e.g., with prefect.docker.DockerImage().build().push() or your CI), then pass the image via job_variables or on the work pool. - Use Flow.from_source(..., entrypoint=...) when deploying from a source If you’re deploying code from a repo or storage, you can specify the entrypoint string there and then call .deploy(). You can still provide a Docker image to run the flow.

from prefect import Flow
from prefect.docker import DockerImage

Flow.from_source(
  source="<https://github.com/acme/flows>",
  entrypoint="src/my_project/impl.py:flow_alpha",
).deploy(
  name="alpha-deploy",
  work_pool_name="docker-pool",
  image=DockerImage(name="<http://ghcr.io/acme/flows:latest|ghcr.io/acme/flows:latest>", dockerfile="Dockerfile"),
)

What’s not available today - Flow.to_deployment(...) does not accept an entrypoint parameter; it infers it. - Flow.deploy(...) does not accept an entrypoint parameter; it infers it. - If you must use Flow.deploy’s Docker build/push convenience but need a different entrypoint, your workaround is to rebind the flow’s underlying function metadata (set flow.fn.__module__/__name__) so the inferred entrypoint matches the symbol you want. References - Deployments overview: Docs: Deployments - Create deployments (Python and CLI): How-to: Create deployments - Docker deployments: How-to: Docker deployments If you tell me which path you prefer (explicit entrypoint with prebuilt image vs. using Flow.deploy’s build/push), I can share a concrete snippet wired for your work pool.

You’re right — the Python deploy you’re looking at (prefect.deployments.deploy or Flow.deploy) does not take an entrypoint parameter. By default, Flow.to_deployment() and Flow.deploy() infer the entrypoint from the flow’s underlying function (module + name). If you want to set the entrypoint explicitly from Python (including for Docker), you have two supported options: 1) Construct a RunnerDeployment with entrypoint and pass it to prefect.deployments.deploy - RunnerDeployment has an entrypoint field. You can set it directly, then deploy and optionally build/push a Docker image.

from prefect.deployments.runner import RunnerDeployment
from prefect.deployments import deploy
from prefect.docker import DockerImage

dep = RunnerDeployment(
    name="alpha-deploy",
    entrypoint="src/my_project/impl.py:flow_alpha",  # explicit entrypoint
    parameters={"x": 2},
    work_pool_name="docker-pool",  # can also be passed to deploy()
)

deploy(
    dep,
    work_pool_name="docker-pool",
    image=DockerImage(name="<http://ghcr.io/acme/flows:latest|ghcr.io/acme/flows:latest>", dockerfile="Dockerfile"),
    build=True,   # build the image
    push=True,    # push the image
)

2) Use Flow.from_source(..., entrypoint=...).deploy(...) - When deploying from a source (e.g., repo or storage), you can set entrypoint on from_source and still provide a Docker image.

from prefect import Flow
from prefect.docker import DockerImage

Flow.from_source(
    source="<https://github.com/acme/flows>",
    entrypoint="src/my_project/impl.py:flow_alpha",
).deploy(
    name="alpha-deploy",
    work_pool_name="docker-pool",
    image=DockerImage(name="<http://ghcr.io/acme/flows:latest|ghcr.io/acme/flows:latest>", dockerfile="Dockerfile"),
)

Notes for Docker - Ensure the entrypoint refers to code that exists and is importable inside the container at runtime. - File form:

./path/to/file.py:function_name

relative to your project root copied into the image. - Module form:

package.module:function_name

if your package is installed in the image. - If you prefer CLI,

prefect deploy ./path/to/file.py:flow_func

allows explicit entrypoint too. Relevant docs - Deployments overview: Docs: Deployments - Create deployments (Python): How-to: Create deployments - Docker deployments: How-to: Docker deployments If you share your repo layout or Dockerfile COPY/WORKDIR, I can suggest the exact entrypoint format that will resolve correctly inside your image.

Great question — there are two different concepts here: - entrypoint: The import target for your flow. It’s the string that points to the callable to run, in either file form or module form: - file form: ./path/to/file.py:flow_func - module form: package.module:flow_func - path: Not a field on RunnerDeployment. “Path” comes from your code source/storage and/or the runtime working directory. It’s the base directory that file-form entrypoints are resolved against. How “path” is determined in common setups - Docker image without storage - You bake your code into the image. There is no storage path on the deployment. - The entrypoint is resolved relative to the container’s working directory at runtime. - Control this by: - Using a module-form entrypoint (package.module:flow) if you install your package in the image (pip install .). - Or using a file-form entrypoint and setting WORKDIR in your Dockerfile so the relative path matches. - Example - Dockerfile sets WORKDIR /app and copies your src/. - entrypoint: src/my_project/impl.py:flow_alpha resolves relative to /app. - Git (or other) storage - You attach a storage object to the deployment. The “path” comes from the storage configuration and where it’s pulled to at runtime. - For GitRepository, Prefect clones the repo into the run environment; your entrypoint is resolved relative to the clone root (or use a module path if the repo is installed as a package). - For LocalStorage(path=...), the entrypoint is resolved relative to that path (mostly useful for local/process workers, not Docker). How to use RunnerDeployment.from_entrypoint with/without storage - Docker image, no storage (code baked into image)

from prefect.deployments.runner import RunnerDeployment
from prefect.deployments import deploy
from prefect.docker import DockerImage

dep = RunnerDeployment.from_entrypoint(
    entrypoint="src/my_project/impl.py:flow_alpha",
    name="alpha-deploy",
    work_pool_name="docker-pool",
    parameters={"x": 2},
)

deploy(
    dep,
    image=DockerImage(name="<http://ghcr.io/acme/flows:latest|ghcr.io/acme/flows:latest>", dockerfile="Dockerfile"),
    build=True,
    push=True,
)
# Ensure your Dockerfile sets WORKDIR and copies src/ so the relative path resolves.

- Git storage + Docker image (clone at runtime)

from prefect.deployments.runner import RunnerDeployment
from prefect.deployments import deploy
from prefect.storage import GitRepository

dep = RunnerDeployment.from_entrypoint(
    entrypoint="src/my_project/impl.py:flow_alpha",
    name="alpha-deploy",
    work_pool_name="docker-pool",
    parameters={"x": 2},
)
dep.storage = GitRepository(
    url="<https://github.com/acme/flows>",
    branch="main",
)

deploy(dep, job_variables={"image": "<http://ghcr.io/acme/runner-base:py310|ghcr.io/acme/runner-base:py310>"})
# Here, the entrypoint is resolved relative to the cloned repo root.

Key takeaways - You cannot set a “path” on RunnerDeployment; there is no such field. The path comes from: - The container’s working directory when you bake code into an image - Or the storage’s base location (e.g., LocalStorage.path, Git clone root) when you attach storage - entrypoint always points to the flow callable; file-form entrypoints are resolved relative to the runtime working directory or storage base; module-form entrypoints are resolved via Python imports. If you share your Dockerfile COPY/WORKDIR and how your code is laid out, I can suggest the exact entrypoint string that will resolve properly in your container, or whether a module-form entrypoint would be simpler.

Short version: use a dotted attribute path in your module-form entrypoint that resolves to a Flow (or plain function). For class “methods,” the safest pattern is to store a Flow object as a class attribute that does not require an instance. Then reference it with module:Class.attr. What works - Define the flow function so it does not expect self/cls, decorate it with @flow, and let it live as a class attribute. Because @flow returns a Flow object (not a descriptor), Python will not try to bind self. Example:

# my_pkg/flows.py
from prefect import flow

class Jobs:
    @flow
    def my_flow(x: int = 1):
        return x * 2

Entrypoint (module form): - my_pkg.flows:Jobs.my_flow This resolves to a Flow object and runs fine. You can verify with:

from prefect.flows import load_flow_from_entrypoint
load_flow_from_entrypoint("my_pkg.flows:Jobs.my_flow")

What to avoid - Don’t stack @classmethod or @staticmethod with @flow. The decorator order won’t produce a usable Flow: - @classmethod returns a classmethod object that @flow cannot decorate - @flow returns a Flow object that @classmethod/@staticmethod cannot wrap - Don’t require self/cls in your flow signature if you want to call it via entrypoint; there will be no instance at runtime. If you really need a “real” static/class method - Keep your implementation as a true static/class method, then expose a thin, top-level flow wrapper that calls it. Reference the wrapper in your entrypoint.

# my_pkg/flows.py
from prefect import flow

class Jobs:
    @staticmethod
    def op(x: int = 1) -> int:
        return x * 2

@flow
def op_flow(x: int = 1):
    return Jobs.op(x)

Entrypoint: my_pkg.flows:op_flow Module vs file entrypoint reminder - module form (what you asked about): my_pkg.flows:Jobs.my_flow - Your package must be importable in the environment (e.g., installed in the Docker image) - file form: src/my_pkg/flows.py:Jobs.my_flow - Resolved relative to the working directory in your container/storage Docs - Deployments overview: Docs: Deployments - Create deployments: How-to: Create deployments If you share a small code sample of your class and your intended entrypoint string, I can confirm the exact attribute path and suggest the cleanest wrapper (if needed).

Thanks for the traceback — that points to a storage/config mismatch rather than the class-method entrypoint itself. What’s happening - You’re using an entrypoint of type module (e.g., my_pkg.flows:Jobs.my_flow). - Your deployment has a storage block attached (likely LocalStorage or another filesystem storage). - When Prefect loads the flow at runtime, it asks the storage to “get_directory(from_path=...)”. For module-form entrypoints, there’s no file path to derive, so from_path is None. Some storage implementations then attempt to copy from “/None”, which triggers your FileNotFoundError. In short: module-form entrypoint + filesystem storage that expects a path = broken. Fixes (pick one) A) Bake code into the Docker image and remove storage - Recommended when you control the image. - Ensure your package is importable in the container, then use a module-form entrypoint. - Do not attach a LocalStorage/RemoteFileSystem storage to the deployment. Example with RunnerDeployment.from_entrypoint:

from prefect.deployments.runner import RunnerDeployment
from prefect.deployments import deploy
from prefect.docker import DockerImage

dep = RunnerDeployment.from_entrypoint(
    entrypoint="my_pkg.flows:Jobs.my_flow",
    name="jobs-my-flow",
    work_pool_name="docker-pool",
    parameters={"x": 2},
)
# ensure no storage is set:
dep.storage = None

deploy(
    dep,
    image=DockerImage(name="<http://ghcr.io/acme/flows:latest|ghcr.io/acme/flows:latest>", dockerfile="Dockerfile"),
    build=True,
    push=True,
)

Dockerfile tips: - Install your package so module imports work:

WORKDIR /app
COPY pyproject.toml poetry.lock ./
RUN pip install .
COPY . .
# or pip install -e .

Validate the entrypoint resolves inside the container:

python -c "from prefect.flows import load_flow_from_entrypoint; load_flow_from_entrypoint('my_pkg.flows:Jobs.my_flow')"

B) Keep storage, but use a file-form entrypoint that resolves to a path in storage - If you need to pull code at runtime (e.g., Git), use storage that supports that and switch to file-form: - entrypoint: "src/my_pkg/flows.py:Jobs.my_flow" - storage: GitRepository(url=..., branch=...) - For filesystem storage, ensure the storage base path actually contains src/my_pkg/flows.py, and use a file-form entrypoint relative to that base. Example with Git:

from prefect.deployments.runner import RunnerDeployment
from prefect.deployments import deploy
from prefect.storage import GitRepository

dep = RunnerDeployment.from_entrypoint(
    entrypoint="src/my_pkg/flows.py:Jobs.my_flow",
    name="jobs-my-flow",
    work_pool_name="docker-pool",
)
dep.storage = GitRepository(url="<https://github.com/acme/flows>", branch="main")

deploy(dep, job_variables={"image": "<http://ghcr.io/acme/runner-base:py310|ghcr.io/acme/runner-base:py310>"})

C) If you must use LocalStorage - Not recommended for Docker, because the container won’t see host paths. - If you do, you must set LocalStorage(path="...") to a directory that exists in the container at runtime (usually it does not), and use a file-form entrypoint relative to that directory. Again, this is fragile with Docker. About class methods with module-form entrypoint - The class attribute must be a Flow object (or a plain function). Avoid needing self/cls. - Example:

# my_pkg/flows.py
from prefect import flow

class Jobs:
    @flow
    def my_flow(x: int = 1):
        return x * 2

Entrypoint: my_pkg.flows:Jobs.my_flow - This works if the package is importable and you’re not forcing filesystem storage that expects a path. Next steps to debug your deployment - Check whether your deployment has a storage attached. If yes and you’re using module-form entrypoint, remove it or switch to Git with a file-form entrypoint. - Confirm your Dockerfile installs your code so my_pkg is importable. - Re-run a quick load check (shown above). Helpful docs - Deployments overview: Deployments

- Create deployments via Python: Create deployments - Docker guidance: Docker deployments - Storage blocks: Storage If you paste your current deployment creation code and Dockerfile COPY/WORKDIR bits, I can suggest the exact minimal change to make your entrypoint resolve cleanly.

You’ve got the entrypoint right; the stack trace points to a storage fetch being attempted even though you expect “no storage.” What’s happening - Your deployment uses a module-form entrypoint and you’ve baked the code into the image. That’s good. - However, at run time Prefect is still trying to “download flow code from storage at None” and calls storage_block.get_directory(...), which indicates storage is configured somewhere for the run. With a module entrypoint and code baked into the image, Prefect should skip storage entirely. - In Prefect 3.x, storage can be attached either on the deployment or at the work pool level as a default. If your work pool has default storage set, the worker will attempt to pull code for every run, even if the deployment doesn’t set storage. That produces from_path=None for module entrypoints and leads to the FileNotFoundError: '/None'. How to confirm - Inspect the work pool to see if default storage is configured:

prefect work-pool inspect "<your-pool-name>"

Look for any “storage” section or hints in the base job template that a “prefect storage pull” step is included. Fixes - Best practice for “module entrypoint + code baked into Docker image”: - Remove default storage from the work pool. In the UI: Work Pools -> your pool -> Storage tab -> clear it. Or recreate the pool without storage. - Keep deployment.storage = None and path = None. - Keep your module-form entrypoint (e.g., my_pkg.flows:Jobs.my_flow). - If you actually want to use storage (Git/S3/etc.), switch to a file-form entrypoint and set a valid path in storage: - Example: entrypoint: "src/my_pkg/flows.py:Jobs.my_flow" and attach GitRepository or other storage that contains src/my_pkg/flows.py at runtime. Optional hardening - Instead of post-editing the entrypoint string, you can build with an explicit module entrypoint:

from prefect.deployments.runner import RunnerDeployment
from prefect.deployments import deploy
from prefect.docker import DockerImage

dep = RunnerDeployment.from_entrypoint(
    entrypoint="my_pkg.flows:Jobs.my_flow",
    name="jobs-my-flow",
    work_pool_name="<your-pool>",
    parameters={"x": 2},
)
# Ensure no deployment-level storage
dep.storage = None

deploy(
    dep,
    image=DockerImage(name="my_image"),
    build=False,
    push=False,
)

Why your “in-container import” test passed but the run failed - Your image is correct and the entrypoint is importable; that’s why load_flow_from_entrypoint works inside the container. - The failure is happening before import: the worker is told to pull code from storage (at None), which crashes first. If you can share the output of: - prefect work-pool inspect "<your-pool-name>" - prefect deployment inspect "<flow-name>/<deployment-name>" I can point to the exact field to change. Also, let me know the pool type and how it was created; some templates add a storage pull step by default if pool storage is configured. References - Deployments: Docs: Deployments - Docker deployments: How-to: Docker deployments - Storage: Concepts: Storage - Work pools: Concepts: Work pools

Thanks for the detailed context — your image and module entrypoint are good; the remaining issue is the work pool’s default storage. What’s going on - Even though your deployment has storage_document_id=None and path=None, your Docker work pool still has a default “storage configuration” attached. In that case, the worker will always try to pull code from the pool’s storage before importing your entrypoint. With a module-form entrypoint, there’s no path to pull, so you see “Downloading flow code from storage at None” and then FileNotFoundError: ‘/None’. Key point: deployment-level storage cannot override/disable a work pool’s default storage. If pool storage is configured, it’s used for all runs on that pool. How to fix it Option A — Clear storage on the work pool (recommended for code-baked images + module entrypoints) - In the UI: Work Pools -> your Docker pool -> Storage tab -> Remove/Clear storage. Save. - After saving, WorkPoolStorageConfiguration() should not show; it should be truly None. - Or recreate a pool with no storage: - CLI:

prefect work-pool create "docker-no-storage" --type docker

- Then point your deployment at this pool:

deploy(deployment, work_pool_name="docker-no-storage", image="my_image", build=False, push=False)

Option B — If you want pool storage, switch to a file-form entrypoint - Attach real storage (e.g., GitRepository) and use a file-based entrypoint relative to the storage root: - entrypoint: "src/my_pkg/flows.py:Jobs.my_flow" - dep.storage = GitRepository(url=..., branch=...) - The worker will clone/pull your code then import the flow. Can I do this at the deployment level in Python? - No. There is no per-deployment flag to “ignore” work-pool storage. If the pool has storage configured, it will be used. To use module entrypoints with code baked into your Docker image, you must remove storage from the work pool or use a pool without storage. Quick checks - Verify the pool has no storage:

prefect work-pool inspect "<your-pool-name>"

Ensure there is no storage block/configuration shown. - Validate the entrypoint inside the image (you already did):

python -c "from prefect.flows import load_flow_from_entrypoint; load_flow_from_entrypoint('my_pkg.flows:Jobs.my_flow')"

If you want, paste the output of: - prefect work-pool inspect "<your-pool-name>" - prefect deployment inspect "<flow>/<deployment>" I’ll confirm the exact field that’s causing the storage pull and provide the minimal change. Helpful docs - Work pools: Concepts: Work pools - Storage: Concepts: Storage - Docker deployments: How-to: Docker deployments

Thanks for digging in — a few concrete answers and steps you can take. What “storage_configuration=WorkPoolStorageConfiguration()” means - That object is the work pool’s storage config wrapper. If it’s truly empty, it should behave as “no storage.” However, given your logs show “Downloading flow code from storage at None,” your worker is still attempting a storage pull. In practice, that happens when the pool has any storage configured or the base job template includes a storage pull step that is firing even when config is empty. Can I change work pool storage via CLI? - Yes. Use the experimental work-pool storage commands: - Inspect current storage:

prefect work-pool storage inspect "<pool-name>"

- Configure storage (S3/GCS/Azure):

prefect work-pool storage configure s3 "<pool-name>" --bucket <bucket> --aws-credentials-block-name <block>

- There is not a “clear storage” subcommand yet. To guarantee no storage: - Create a new pool without storage and point your deployments to it:

prefect work-pool create "docker-no-storage" --type docker

- Or overwrite an existing pool’s base job template to remove the storage pull step (see below). Can a deployment-level setting disable pool storage? - No. Deployment-level settings cannot override a work pool’s storage. If the pool has storage configured or its job template always executes a storage pull step, the worker will try to pull code regardless of your deployment’s storage=None and path=None. Two reliable fixes A) Use a pool with no storage and a job template with no storage pull step - Create a new pool (or overwrite an existing pool) and ensure its base job template does not include a storage step. - Get the default template, edit it, and update: 1) Export the default Docker pool job template:

prefect work-pool get-default-base-job-template --type docker --file docker-template.json

2) Open docker-template.json and remove any step that references storage pulls (e.g., a step whose command includes “prefect storage …”). 3) Apply the edited template:

prefect work-pool create "docker-no-storage" --type docker --base-job-template docker-template.json --overwrite

- Now, keep deployment.storage=None and module-form entrypoints. Since your image test succeeded, this should stop the “Downloading flow code from storage at None” error. B) If you actually want pool storage, switch to a file-form entrypoint - Attach real storage (e.g., GitRepository) and set a file-based entrypoint relative to the storage root: - entrypoint:

src/my_pkg/flows.py:Jobs.my_flow

- configure storage on the pool with the CLI or attach storage on the deployment. Programmatic update (Python) - There isn’t a public Python helper to “clear” pool storage. You can: - Create a new pool with no storage via CLI (simplest). - Or use the orchestration client to update the work pool’s base job template and/or storage configuration. If you want this route, share your Prefect version and I’ll give you an exact snippet that sets the pool’s storage configuration to None and updates the job template. Sanity checks - Inspect what Prefect thinks:

prefect work-pool inspect "<pool-name>"
  prefect work-pool storage inspect "<pool-name>"

- Re-run your deployment with the storage-less pool: - Make sure you point deploy() to that pool:

deploy(deployment, work_pool_name="docker-no-storage", image="my_image", build=False, push=False)

Links - Work pools: Concepts: Work pools - Storage: Concepts: Storage - Docker deployments: How-to: Docker deployments If you paste the output of: -

prefect work-pool storage inspect "<your-pool-name>"

-

prefect work-pool inspect "<your-pool-name>"

I’ll call out the exact field that’s triggering the storage pull and, if needed, provide a ready-to-run Python snippet to update the pool for your Prefect version.

Got it — thanks for checking. Since work-pool storage inspect shows “No storage configuration found”, the pool itself isn’t pointing to S3/GCS/etc. Yet your flow still attempts a storage pull and crashes on “/None”. That usually means the runtime is still constructing a filesystem storage object with a None path. Practical ways to get unblocked 1) Force “no-op” storage on the deployment Attach a NullFileSystem to the deployment so any storage pull becomes a no-op instead of a LocalFileSystem copy from /None.

from prefect.filesystems import NullFileSystem
from prefect.deployments import deploy

dep = my_flow.to_deployment(..., entrypoint_type=EntrypointType.MODULE_PATH)
dep.entrypoint = f"{my_flow.__module__}:{my_flow.__qualname__}"
dep.storage = NullFileSystem()  # explicitly disable pulling code

deploy(
  dep,
  work_pool_name="docker-no-storage",
  image="my_image",
  build=False,
  push=False,
)

2) Double-check the pool’s base job template Even with “no storage,” a custom base job template can add a pull step. If you’ve customized the Docker pool template, export it and ensure it doesn’t include any storage/pull commands: - Export default template for docker:

prefect work-pool get-default-base-job-template --type docker --file docker.json

- If you’ve got a custom template, remove any storage/pull step and re-apply:

prefect work-pool create "docker-no-storage" --type docker --base-job-template docker.json --overwrite

3) Verify deployment definition on the server Make sure the server really sees storage=None and entrypoint as module-form:

prefect deployment inspect "<flow-name>/<deployment-name>"

4) Keep Prefect versions aligned and current Make sure the Prefect version used to build the image matches your worker and server/Cloud version. If you’re not on the latest 3.x, please upgrade in both your image and worker environment — we’ve fixed a few edge cases here. About “updating work pool storage” via CLI/Python - CLI: You can inspect and configure pool storage with the experimental commands: - Inspect:

prefect work-pool storage inspect "<pool-name>"

- Configure (e.g., S3):

prefect work-pool storage configure s3 "<pool-name>" --bucket ... --aws-credentials-block-name ...

- There isn’t a “clear storage” command; to guarantee no storage, create a new pool without configuring storage or overwrite the pool with a base job template that does not pull code. - Python: There isn’t a stable public SDK helper to clear pool storage. You can script pool creation with CLI as above, or I can share a minimal client snippet for your Prefect version if you want to automate it. If the NullFileSystem workaround resolves the error, that confirms something in your runtime was constructing a filesystem even with “no storage” configured. If it doesn’t, please share: - Prefect version (client/image and worker) - Output of

prefect deployment inspect "<flow>/<deployment>"

- Whether the Docker pool uses a custom base job template I’ll tailor the next step from there.

That ValidationError makes sense: RunnerDeployment.storage expects a RunnerStorage, not a Filesystem block. NullFileSystem is a filesystem block; it won’t validate there. What this means - storage on a deployment must be one of the RunnerStorage types in prefect.storage (e.g., GitRepository, RemoteStorage, LocalStorage, BlockStorageAdapter), not a filesystem block like NullFileSystem or LocalFileSystem. - Since you’re baking code into the image and using a module-form entrypoint, the correct setup is actually storage=None and a work pool that does not attempt any storage pull step. Why you still see “Downloading flow code from storage at None” - Even with “no storage configuration found” on the pool, the pool’s base job template can still include a storage pull step that runs unconditionally. That will create a “storage” context and try to pull using a None path, yielding the /None error. Two reliable ways to fix A) Remove any storage pull step from the Docker pool’s base job template - Export the default template, edit it, and update the pool. 1) Export:

prefect work-pool get-default-base-job-template --type docker --file docker.json

2) Open docker.json and remove any step that pulls from storage (e.g., a step whose command runs something like “prefect storage …” or otherwise references storage). Keep only the step that runs the flow. 3) Apply to your pool:

prefect work-pool update "<your-docker-pool>" --base-job-template docker.json

Now keep your deployment with: - entrypoint = "my_pkg.flows:Jobs.my_flow" - storage = None - path = None B) If you do want pool-level storage, switch to a file-form entrypoint and real storage - Attach storage at the pool or deployment, and use a file-based entrypoint relative to that storage. - Example: Git + file-form entrypoint

from prefect.deployments.runner import RunnerDeployment
from prefect.deployments import deploy
from prefect.storage import GitRepository

dep = RunnerDeployment.from_entrypoint(
    entrypoint="src/my_pkg/flows.py:Jobs.my_flow",
    name="jobs-my-flow",
    work_pool_name="docker-pool",
)
dep.storage = GitRepository(url="<https://github.com/acme/flows>", branch="main")

deploy(dep, job_variables={"image": "<http://ghcr.io/acme/runner-base:py310|ghcr.io/acme/runner-base:py310>"})

About setting a “no-op” storage on the deployment - Don’t use NullFileSystem; it’s a filesystem block, not a RunnerStorage. - You could set RunnerStorage.LocalStorage(path=".") but that still triggers a copy step and may not be what you want. The clean approach for code-baked images is: storage=None and a pool template with no storage pull step. Quick checklist - Your image can import the entrypoint (you’ve verified). - Your deployment shows storage_document_id=None and path=None (good). - Update the Docker pool’s base job template to remove any storage pull step (most likely root cause here). - Re-run and confirm the “Downloading flow code from storage at None” line is gone. If you want, paste your current pool’s base job template (export it with get-default-base-job-template or from work-pool inspect) and I’ll point out exactly what to remove. Also, which Prefect version are your worker and image on? If you’re not on latest 3.x, upgrading is recommended as we’ve improved this path recently. Docs - Work pools: Work pools - Storage: Storage - Docker deployments: Docker deployments