Short answer: that’s expected. In Prefect 3.x,

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

pulls and imports your flow code on the machine where you run the deploy command.

PREFECT_API_URL

only tells the SDK which API to talk to; it does not change where code is imported during deployment. What happens: - During deploy, Prefect clones/pulls the source locally and imports the flow to extract metadata and create the deployment. So the deploy machine must be able to access your repo and import your flow (including its Python dependencies). - At run time, the worker reads the deployment’s pull steps from the API, pulls the code to the worker, then imports and executes the flow there. The server never hosts your code. How to proceed: - Run the deploy step in an environment that can access and import your flow code (e.g., your dev machine with the repo and dependencies, or your CI runner). - Or run the deploy command on the worker itself. - Or push your code to a Git repo and use:

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

Ensure the deploy machine can clone the repo and import

flows.py

. - If you’re baking code into a container image for your worker, you still need to create the deployment from an environment that can import the flow, but at runtime the worker will use the image (no extra pull steps needed if the code is baked in and entrypoint resolves inside the container). A couple of checks: - Make sure your

entrypoint

is correct and relative to the repo root (e.g.,

src/flows.py:my_flow

). - Confirm the machine running deploy has credentials to pull the repo and can import top-level dependencies of the flow. If you share your exact deploy snippet and the error/trace you’re seeing, I can pinpoint the fix. For background on deployments and pull steps in Prefect 3.x, see the docs: Prefect 3.x docs.

You’re hitting this because your

_source_

is a local path on the deploy machine. In Prefect 3.x, a local

source

creates a deployment that points to that same path at runtime. The worker then tries to import the flow from that path on the worker host, which doesn’t exist—hence it “looks for your client path” and fails. Why: -

afrom_source(_source_=settings.working_dir, ...)

uses LocalStorage. LocalStorage does not transfer code; it records a path. - At runtime the worker sets its working directory to that recorded path and imports your entrypoint. If the path doesn’t exist on the worker, import fails. Fix options: 1) Use a pullable source (recommended) - Put the code in a Git repo and use that as

source

. The worker will clone the repo via pull steps at runtime.

to_deploy = await dag_runner.afrom_source(
    _source_="<https://github.com/<org>/<repo>.git>",   # optionally add @branch or use GitRepository for creds
    _entrypoint_="app/workflow/prefect/flows/dag_runner.py:dag_runner",
)
deployment_uuid = await to_deploy.adeploy(
    _name_=f"ks-workflow-instance-{_input_.workflow_instance_code}",
    _work_pool_name_=settings.prefect_dag_worker_pool,
    _parameters_={"workflow_input": _input_.model_dump()},
)

2) Bake or install the code in the worker image/env - Omit a pullable source and make the entrypoint importable in the worker’s Python environment (e.g., pre-baked Docker/K8s image or installed package). Run the deploy step in an environment that can import the flow as well. 3) Keep LocalStorage but share the same path - Ensure the exact same directory and code exist on the worker at

settings.working_dir

(e.g., mount a shared volume or mirror the path). This couples your deployment to that filesystem layout. A few tips: - Your entrypoint looks fine as a file path:

app/workflow/prefect/flows/dag_runner.py:dag_runner

- If you switch to Git, make sure the worker has credentials to clone the repo at runtime. - If you keep LocalStorage, confirm the worker can access that path and that the file exists there. 中文总结： - 你现在用的是本地路径作为 `source`，部署记录了这个路径，运行时 worker 会到这个路径导入 flow。如果 worker 上没有这个路径，就会报错。 - 解决方案：改成 Git 源码仓库作为 `source`（worker 运行时会自动拉代码）；或者把代码打进 worker 的镜像/环境；或者在 worker 上挂载相同路径的代码目录。 If you can share your worker type (process/docker/k8s) and whether you can use Git or a container image, I can give a concrete example for your setup. Docs for background: https://docs.prefect.io/latest