Got it — you want to move from local-process runs to Kubernetes workers on a self-hosted Prefect Server, and you’d like to be able to run

.deploy()

from any of four servers. Below is a proven path in Prefect 3.x that supports your setup. High-level plan - Run Prefect Server (since you’re not using Cloud). - Create a Kubernetes work pool. - Start a Kubernetes worker that can talk to your cluster (inside or outside the cluster). - Change your deployments to use a shareable source of code (recommended: Git) or build/push an image that includes your code. - Then any of your 4 servers can run

.deploy()

safely. 1) Run Prefect Server (self‑hosted) - Start the API/UI locally or on a VM reachable by your worker:

prefect server start

- On all 4 servers (and anywhere you run

prefect

CLI or

.deploy()

), point to your Server:

export PREFECT_API_URL="http://<your-server-host>:4200/api"
prefect profile inspect

2) Install Kubernetes support and create a Kubernetes work pool - On machines that will run the worker and/or `.deploy()`:

pip install "prefect[kubernetes]"

- Create a Kubernetes work pool (verified command):

prefect work-pool create "k8s-pool" --type kubernetes

- (Optional) Pull and customize the base job template (namespace, SA, imagePullSecrets, resources, env):

prefect work-pool get-default-base-job-template --type kubernetes --file base-job-template.json
# edit base-job-template.json, then:
prefect work-pool update "k8s-pool" --base-job-template base-job-template.json

3) Start a Kubernetes worker You can run the worker either outside or inside the cluster: - Outside the cluster (uses your kubeconfig): - Ensure

kubectl get nodes

works from the worker host (kubeconfig and RBAC in place). - Start the worker:

prefect worker start --pool "k8s-pool"

- Inside the cluster (as a Deployment/Helm): - Create a ServiceAccount/Role to allow creating Jobs/reading Pods. - Run a Deployment that runs

prefect worker start --pool k8s-pool

and set

PREFECT_API_URL

so the worker can reach your Server. - Helm chart (for convenience): Prefect Helm charts Minimal RBAC (adjust namespace names as needed):

apiVersion: v1
kind: ServiceAccount
metadata:
  name: prefect-worker
  namespace: prefect
---
apiVersion: <http://rbac.authorization.k8s.io/v1|rbac.authorization.k8s.io/v1>
kind: ClusterRole
metadata:
  name: prefect-worker-role
rules:
  - apiGroups: ["batch"]
    resources: ["jobs"]
    verbs: ["create","get","list","watch","patch","delete"]
  - apiGroups: [""]
    resources: ["pods","pods/log","pods/status"]
    verbs: ["get","list","watch"]
---
apiVersion: <http://rbac.authorization.k8s.io/v1|rbac.authorization.k8s.io/v1>
kind: ClusterRoleBinding
metadata:
  name: prefect-worker-binding
roleRef:
  apiGroup: <http://rbac.authorization.k8s.io|rbac.authorization.k8s.io>
  kind: ClusterRole
  name: prefect-worker-role
subjects:
  - kind: ServiceAccount
    name: prefect-worker
    namespace: prefect

4) Choose a deployment pattern that works from any of the 4 servers To avoid the “local path” problem (the Kubernetes job won’t see your local filesystem), use one of these: Option A — Git source + base image (recommended) - Put your flows in a shared Git repo. - Use an image that already has your Python dependencies (no code baked in). The worker will pull code from Git at runtime. - Deploy from any server: ``` from prefect import flow @flow def my_flow(): print("hello from k8s") if name == "__main__": flow.from_source( source="https://github.com/your-org/your-repo.git", entrypoint="flows/my_flow.py:my_flow", ).deploy( name="my-deployment", work_pool_name="k8s-pool", image="your-registry/prefect-base:py3.11", # deps baked-in # optional: target a specific queue or set job vars work_queue_name="default", job_variables={ "namespace": "prefect", # include service account or image pull secrets if needed

# "service_account_name": "prefect-worker", }, tags=["k8s","prod"], )

Notes:
- The image should contain your dependencies (requirements.txt installed). Your flow code is pulled from the Git repo at run time.
- You can safely run this `.deploy()` from any of the 4 servers; it will update the same deployment by name.

Option B — Build/push an image that contains your code (keeps local path viable)
- Keep `flow.from_source(source=<local_path>, ...)` if you want, but always build/push an image during deploy so the code is baked into the runtime container.
- Any of the 4 servers can run deploy as long as they can build/push to your registry.

from pathlib import Path from prefect import flow @flow def my_flow(): print("code is baked into the image") if name == "__main__": flow.from_source( source=str(Path(file).parent), # local path OK if you build the image entrypoint="flows/my_flow.py:my_flow", ).deploy( name="my-deployment", work_pool_name="k8s-pool", image="your-registry/your-app:my-tag", build=True, # build Docker image with your code push=True, # push the image to your registry job_variables={"namespace": "prefect"}, )

Notes:
- This approach avoids Git, but you must have a registry and consistent image tags/versioning.
- If two servers deploy the same name concurrently with different code, the last pushed image “wins”.

5) Trigger runs and iterate
- After deploying:

prefect deployment ls prefect deployment run "my_flow/my-deployment"

- To route by priority, use work queues:
  - Start worker with queues:

prefect worker start --pool "k8s-pool" --work-queue "high" --work-queue "default"

- Deploy to a queue:

my_flow.deploy(..., work_queue_name="high")

Practical tips
- Always install extras where needed: `pip install "prefect[kubernetes]"`.
- Prefer Git source for multi-machine deployments; it’s the simplest way to guarantee code consistency.
- If you use a private registry or repo:
  - Add `imagePullSecrets` to your base job template.
  - Use tokens/SSH keys for Git in your deployment’s pull step, or bake code into the image.
- Make sure your worker process has permissions and network to reach both:
  - Your Prefect Server `PREFECT_API_URL`
  - The Kubernetes API

Validated CLI
- Work pool create:

prefect work-pool create --help

- Worker start:

prefect worker start --help

- Server start:

prefect server start --help ``` A couple of questions to tailor this further - Do you prefer the Git-pull-at-runtime approach or baking code into Docker images? - Will the Kubernetes worker run inside the cluster (Deployment) or on one of your 4 servers using kubeconfig? - Do you have a Docker registry available for images? If you share your preferred pattern and your cluster/registry constraints, I can provide a minimal, copy‑pasteable configuration (job template, Deployment/Helm values, and a deploy script) for your environment.

Great questions. Short answers first: - If your code is baked into a Docker image, you do not need

flow.from_source

. That’s only for pulling code at runtime (e.g., from Git/S3). With a baked image, Prefect just needs to know how to import your flow inside the container. - Yes, you can reuse a single preset Docker image for many deployments that all use the same flow function with different parameters. You do not need to rebuild per deployment. Rebuild only when your code or dependencies change. How it works with a baked image - Prefect records an “entrypoint” for your flow (either a file path inside the container, or a module import path). When a run starts, the Kubernetes job uses the image you specify and imports your flow via that entrypoint. - No

from_source

needed because the code is already in the container. Recommended setup 1) Make sure your Docker image contains: - Your flow code in a consistent location - All Python dependencies Example Dockerfile:

FROM prefecthq/prefect:3-latest
WORKDIR /app
COPY . /app
RUN pip install -r requirements.txt

2) Build/push once (outside Prefect), then reuse the tag:

docker build -t your-registry/your-image:1.0.0 .
docker push your-registry/your-image:1.0.0

3) Create multiple deployments reusing the same image (different parameters) - You can deploy from any of your 4 servers. Just point to the same image and skip building:

from prefect import flow, deploy

@flow
def my_flow(env: str):
    print(f"Running for {env}")

if __name__ == "__main__":
    deploy(
        my_flow.to_deployment(
            name="flow-us",
            parameters={"env": "us"},
        ),
        my_flow.to_deployment(
            name="flow-eu",
            parameters={"env": "eu"},
        ),
        my_flow.to_deployment(
            name="flow-apac",
            parameters={"env": "apac"},
        ),
        work_pool_name="k8s-pool",
        image="your-registry/your-image:1.0.0",
        build=False,   # reuse existing image
        push=False,    # reuse existing image
    )

Notes and tips - If Prefect can find your flow locally when you call

.deploy()

, it will record the entrypoint automatically. Ensure the same relative path exists inside the image (the Dockerfile above sets

WORKDIR /app

and

COPY . /app

, which matches typical repo layouts). - Prefer versioned image tags (e.g.,

1.0.0

) instead of

latest

so worker pods don’t accidentally pick up new code. - If you package your code as a Python module inside the image, you can use a module import entrypoint (instead of a file path). That’s a nice way to decouple from filesystem paths, but not required. - You only need to rebuild the image when your code or dependencies change. Creating more deployments that reference the same flow with different parameters does not require a rebuild. Triggering with different parameters at run time (optional) - You can still override parameters per run without rebuilding:

prefect deployment run "my_flow/flow-us" -p env=us-adhoc

If you want, I can help you: - Confirm your entrypoint lines up with your image layout, or - Switch to a module-path entrypoint to make things fully import-based, or - Provide a minimal CI job to build/push your image once, then have your 4 servers run

.deploy()

with

build=False, push=False

. For Helm (if you want to run the worker in-cluster), see Prefect Helm charts.

Short answer: the image you put in the YAML that starts the Kubernetes worker only sets the worker’s own container image. It does not control the image used to run your flow. The image for flow runs is set on the work pool/deployment side and is passed to the worker at run time. What to do instead: - Set the flow-run image via deployment

job_variables

(highest priority), or as a default on the Kubernetes work pool (applies to all deployments in that pool). - The worker will render the base job template using those values and create a Kubernetes Job with the specified image. Example with

.deploy()

in Prefect 3.x:

from prefect import flow

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

my_flow.deploy(
    name="prod",
    work_pool_name="my-k8s-pool",
    job_variables={
        "image": "<http://ghcr.io/acme/my-custom-flow-image:1.2.3|ghcr.io/acme/my-custom-flow-image:1.2.3>",  # image used by the Job/pod
        "image_pull_policy": "IfNotPresent",                 # optional
        "namespace": "prefect"                               # optional
    },
    parameters={"x": 5}  # deployment default parameters
)

Alternatively, set a default image on the work pool: - In the UI: Work Pools -> your Kubernetes pool -> Variables -> set

image: <http://ghcr.io/acme/my-custom-flow-image:1.2.3|ghcr.io/acme/my-custom-flow-image:1.2.3>

- Then omit

job_variables

from individual deployments unless you need an override. About parameters at scheduled time: - The

parameters

you set on the deployment are stored with the deployment. - When a schedule triggers, Prefect Server/Cloud creates a flow run that includes those parameters. - The worker pulls the run, renders the job, and the flow executes with those parameters—no need to put parameters in the worker YAML. Key distinctions: - Worker YAML image = the worker pod itself (polling/creating Jobs). Changing this does not change the image used by flow runs. - Flow run image = set via deployment

job_variables["image"]

or work pool variables; this is the image your flow code runs in. If you need to pull from a private registry, add

imagePullSecrets

in the work pool’s base job template or include it in your job template so the worker-created Jobs can authenticate.

Short answer: yes. You can deploy with

Flow.from_source(...)

to a Process work pool, and you can run a Process worker anywhere (including in a Kubernetes pod) that polls that pool. The code does not need to be present where the worker runs—Prefect will fetch it from the repo at run time via the deployment’s pull steps. Just make sure the worker type matches the pool type. Key points: - Use

Flow.from_source(source=<git url>, entrypoint="path/to/file.py:flow_func").deploy(...)

. The deployment will store “pull steps” to fetch your repo when a run starts. - Start a Process worker (not a Kubernetes worker) and point it at your Process pool. You can run this worker inside Kubernetes; it will still execute flows as local subprocesses inside that pod. - Worker type must match pool type: - Process worker → Process pool (works) - Kubernetes worker → Kubernetes pool (works) - Kubernetes worker → Process pool (won’t work) - Ensure the worker environment has: - Network access to Prefect API and your Git repo - Credentials to clone the repo (PAT/SSH, secrets) - Python and all your flow’s dependencies installed (bake them into the worker image or handle them at startup) Example deployment with GitHub source (Prefect 3.x):

from prefect import flow

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

# Point at your repo and the flow entrypoint in that repo
Flow.from_source(
    source="<https://github.com/acme/data-platform.git>",
    entrypoint="flows/my_flow.py:my_flow",
).deploy(
    name="prod",
    work_pool_name="my-process-pool",
    parameters={"x": 5},
)

Start a Process worker (can be in a Kubernetes pod/container):

prefect worker start --pool my-process-pool

Notes: - If your repo is private, provide credentials (e.g., use a PAT in the URL or mount env vars/secrets into the worker pod). - For dependencies, either: - Build them into the worker image that runs the Process worker; or - Ensure they can be installed at startup before the worker begins polling. Docs: - Deploy via Python (from_source) - Store flow code (Git, etc.) - Work pools - Workers If you want, I can share a minimal Kubernetes manifest for running a Process worker in a pod (with the right env vars and command) based on your environment (Prefect Cloud vs. Server, image name, and repo access).