Short answer: because in Prefect 3.x your workers typically run your flow in a different environment than your local machine, they need a way to fetch your code. Prefect requires one of: - a Docker image that already contains your code (pass

image=...

), or - a remote source for your code (use

flow.from_source(...)

), or - a module-path entrypoint where the code is already installed on the worker Without one of those, the worker can’t import your local script at run time, so Prefect raises: "Either an image or remote storage location must be provided when deploying a deployment." How to fix it (pick one): 1) No Docker, use a Git repo (recommended simple path) - Put your flow script in a repo and deploy from it:

from prefect import flow, task

@task
def get_customer_ids() -> list[str]:
    return [f"customer{n}" for n in range(10)]

@task
def process_customer(customer_id: str) -> str:
    return f"Processed {customer_id}"

@flow
def main() -> list[str]:
    ids = get_customer_ids()
    # In Prefect 3.x, use submit for parallel task runs
    futs = [process_customer.submit(cid) for cid in ids]
    return [f.result() for f in futs]

if __name__ == "__main__":
    main.from_source(
        source="<https://github.com/<org>/<repo>.git>",  # or S3/GS/etc.
        entrypoint="path/to/your_script.py:main",
    ).deploy(
        name="example-deployment",
        work_pool_name="default",
    )

2) Use a Docker image - Let Prefect build/push an image that includes your code and dependencies:

if __name__ == "__main__":
    main.deploy(
        name="example-deployment",
        work_pool_name="default",
        image="<http://ghcr.io/<org>/<image>:latest|ghcr.io/<org>/<image>:latest>",  # your registry/image
        build=True,
        push=True,
    )

3) Same machine/process worker, no image or remote storage - If your worker runs on the same machine and your code is installed/importable there, you can use a module-path entrypoint:

if __name__ == "__main__":
    main.deploy(
        name="example-deployment",
        work_pool_name="default",
        entrypoint_type="module_path",  # code must be importable on the worker
    )

For this to work, ensure the worker’s environment can import your flow (e.g.,

pip install -e .

or

PYTHONPATH

set appropriately). Notes about your snippet: -

process_customer.map(...)

is a Prefect 2.x pattern. In Prefect 3.x use

task.submit(...)

inside a list comprehension and gather results with

.result()

as shown above. If you tell me which work pool you’re using (Docker, Kubernetes, Process, etc.) and whether you want to avoid Docker, I can provide the exact deployment snippet for your setup.