Short answer: it depends what you mean by “multiple flow runs.” - If you mean multiple flows in one deployment: No. In Prefect 2.20.6, each deployment in prefect.yaml points to a single flow via

entrypoint

. To manage multiple flows, define multiple deployments (one per flow) in the

deployments:

list. - If you mean creating many runs of the same flow from one deployment: Yes. You can attach multiple schedules to a single deployment. Each schedule will create its own stream of runs. Notes specific to 2.20.6: - Multiple schedules per deployment are supported. - Per-schedule parameter overrides are not supported in 2.20.6; all schedules use the deployment-level

parameters

. If you need different params, create separate deployments (or upgrade to a later 2.x version that adds per-schedule params). Example prefect.yaml (2.20.6):

deployments:
  - name: my-flow-deployment
    entrypoint: flows/my_flow.py:my_flow
    parameters:
      mode: "standard"
    schedules:
      - cron: "0 6 * * *"         # daily at 06:00
        timezone: "UTC"
        active: true
      - interval: 3600            # every hour
        active: true

  # Another deployment for a different flow or different params
  - name: my-flow-deployment-fast
    entrypoint: flows/my_flow.py:my_flow
    parameters:
      mode: "fast"
    schedules:
      - cron: "0 12 * * *"
        timezone: "UTC"

If you were hoping to target multiple flows with a single deployment, the supported pattern is to add one deployment entry per flow in the same prefect.yaml.

Great question! Here’s the clear separation between the two in Prefect 2.x (e.g., 2.20.6): What each one is - prefect.yaml - A project-level file you commit to your repo. - Defines how to build/push/pull your code and the deployments you want to register (flow entrypoints, parameters, schedules, tags, which work pool to use, and any per-deployment job_variables). - Consumed by prefect deploy. - Work pool job template - A pool-level configuration stored in the Prefect API (Cloud/Server), created/edited via UI or CLI. - Defines which infrastructure variables exist for that pool type (schema) and their defaults (e.g., image, env vars, resources, stream_output). - Used by workers when launching flow runs from that pool; deployments supply overrides via job_variables that fill into this template. Scope and ownership - prefect.yaml - Scope: a single project/repo; can define multiple deployments. - Owned by: flow developers/data engineers/DevOps on the app team. - Lives: in your repo (default path ./prefect.yaml). - Work pool job template - Scope: one work pool; shared by all deployments targeting that pool. - Owned by: platform/infra team (governance and defaults). - Lives: in Prefect (view/edit in UI or via CLI; not typically in your repo). When they’re used - prefect.yaml is read during prefect deploy to: - Run build/push steps (e.g., build a Docker image). - Register deployments and attach them to a work pool. - The job template is applied at run submission time by the worker: - Worker merges the pool template defaults with deployment/job-level overrides to create the final runtime configuration. Variable precedence (highest → lowest) 1) Runtime job variables (e.g., set via UI/CLI when starting a run) 2) Deployment job_variables (defined in prefect.yaml under deployments[].work_pool.job_variables) 3) Work pool job template defaults (defined by the pool) Simple examples - prefect.yaml

name: my-data-pipeline
prefect-version: 2.20.6

deployments:
  - name: prod-etl
    entrypoint: flows/etl.py:run_etl
    parameters:
      environment: production
    schedules:
      - cron: "0 2 * * *"
        timezone: UTC
    work_pool:
      name: prod-docker-pool
      job_variables:
        image: "my-registry/etl:latest"
        env:
          LOG_LEVEL: INFO

- Work pool job template (conceptual snippet; stored with the pool)

{
  "job_configuration": {
    "image": "{{ image }}",
    "env": "{{ env }}",
    "stream_output": "{{ stream_output }}"
  },
  "variables": {
    "type": "object",
    "properties": {
      "image": {
        "type": "string",
        "default": "prefecthq/prefect:2-latest"
      },
      "env": {
        "type": "object",
        "default": {}
      },
      "stream_output": {
        "type": "boolean",
        "default": true
      }
    }
  }
}

With the above: - If the deployment sets job_variables.image to my-registry/etl:latest, that overrides the pool’s default image. - If nothing overrides stream_output, it stays true from the pool default. - If you start a run and specify a job variable at runtime, that wins over both. Helpful docs - Work pools overview: Work Pools & Workers - Prefect deploy and prefect.yaml: Deploy with prefect.yaml - Overriding job variables and precedence: Overriding Work Pool Job Variables - Work pool CLI reference: prefect work-pool CLI

Here’s a solid starting point for a prefect.yaml that deploys to a Kubernetes-based worker/work pool in Prefect 2.20.6, plus notes on the common job_variables you can set. Key ideas - You create a Kubernetes work pool (via UI or CLI), then point your deployments at it. - Pool-level defaults come from the pool’s base job template; you override them per deployment via work_pool.job_variables in prefect.yaml. - Commonly used job_variables for k8s: image, namespace, service_account_name, image_pull_policy, env, labels, finished_job_ttl, job_watch_timeout_seconds, pod_watch_timeout_seconds, stream_output. Example prefect.yaml (Kubernetes)

name: my-k8s-project
prefect-version: 2.20.6

# Optionally build and push an image used by your k8s jobs
build:
  - prefect_docker.deployments.steps.build_docker_image:
      id: build-image
      requires: prefect-docker>=0.4.0
      image_name: "my-registry/my-flow-image"
      tag: "latest"
      dockerfile: auto
      platform: "linux/amd64"
      push: true

# Optional: if you run from a Git checkout at runtime
pull:
  - prefect.deployments.steps.git_clone:
      repository: <https://github.com/my-org/my-repo.git>
      branch: main
      # directory: flows  # set if your flows live in a subdir

deployments:
  - name: my-flow-prod
    entrypoint: flows/my_flow.py:my_flow
    parameters:
      mode: "prod"
    schedule:
      cron: "0 9 * * *"
      timezone: UTC
    work_pool:
      name: my-k8s-pool           # must match an existing Kubernetes work pool
      # Per-deployment overrides of the pool’s job template variables:
      job_variables:
        image: "{{ build-image.image }}"     # or a fixed image like "my-registry/my-flow-image:latest"
        namespace: "prefect"
        service_account_name: "prefect-worker"
        image_pull_policy: "IfNotPresent"    # Allowed: IfNotPresent, Always, Never
        finished_job_ttl: 3600               # seconds to retain completed Jobs
        job_watch_timeout_seconds: 7200      # how long the worker waits on Job events
        pod_watch_timeout_seconds: 300       # how long to wait for a Pod to appear
        stream_output: true
        env:
          LOG_LEVEL: INFO
          EXTRA_PIP_PACKAGES: "pandas==1.5.3"
        labels:
          team: data-platform
          environment: production

  - name: my-flow-dev
    entrypoint: flows/my_flow.py:my_flow
    parameters:
      mode: "dev"
    work_pool:
      name: my-k8s-pool
      job_variables:
        image: "{{ build-image.image }}"
        namespace: "prefect-dev"
        service_account_name: "prefect-worker"
        image_pull_policy: "IfNotPresent"
        finished_job_ttl: 600
        stream_output: true
        env:
          LOG_LEVEL: DEBUG
        labels:
          team: data-platform
          environment: development

Common Kubernetes job_variables you can set on deployments - image: container image for the flow run (e.g., docker.io/prefecthq/prefect:2-latest or your custom build). - namespace: Kubernetes namespace where the Job will run (default “default” if pool/template sets it that way). - service_account_name: service account name used by the Job/Pod. - image_pull_policy: IfNotPresent, Always, or Never. - finished_job_ttl: seconds to keep a completed Job before K8s garbage-collects it. - job_watch_timeout_seconds: how long the worker watches for Job events before timing out. - pod_watch_timeout_seconds: how long to wait for a Pod to appear. - stream_output: whether to stream logs to the worker’s stdout. - env: dict of environment variables injected into the container. - labels: dict of labels applied to the Job/Pod via the base job template. Tips - Create your pool first (UI: Work Pools → New → Kubernetes) and keep its base job template mostly “standard,” then override per deployment via job_variables.

- If you need CPU/memory requests/limits or custom volumes, add variables and wiring in the pool’s base job template (Advanced tab in the pool) and then set them via deployment job_variables. - The final runtime config is: pool template defaults < deployment job_variables < runtime job variable overrides (e.g., via UI/CLI when starting a run). If you share your target image/namespace/service account, I can tailor the example exactly to your setup.

Here’s a minimal, working way to create a Process-type work pool via Python in Prefect 2.20.6. Minimal example (2.20.6, async client)

import asyncio
from prefect.client.orchestration import get_client
from prefect.client.schemas.actions import WorkPoolCreate
from prefect.workers.process import ProcessWorker

async def main():
    # Get the default base job template for a process worker
    base_job_template = ProcessWorker.get_default_base_job_template()

    async with get_client() as client:
        work_pool = await client.create_work_pool(
            work_pool=WorkPoolCreate(
                name="my-process-pool",
                type="process",
                base_job_template=base_job_template,
                is_paused=False,           # optional
                description="Created via Python",  # optional
                # concurrency_limit=10,     # optional
            )
        )
        print(f"Created work pool: {work_pool.name}")

asyncio.run(main())

Customizing defaults in the base job template (e.g., env, working_dir, stream_output)

import asyncio
from prefect.client.orchestration import get_client
from prefect.client.schemas.actions import WorkPoolCreate
from prefect.workers.process import ProcessWorker

async def main():
    base_job_template = ProcessWorker.get_default_base_job_template()

    # Set defaults that all deployments on this pool will inherit (unless overridden)
    base_job_template["job_configuration"]["env"] = {
        "PREFECT_LOGGING_LEVEL": "INFO",
        "MY_CUSTOM_VAR": "value"
    }
    base_job_template["variables"]["properties"]["stream_output"]["default"] = False
    base_job_template["variables"]["properties"]["working_dir"]["default"] = "/var/prefect/runs"

    async with get_client() as client:
        work_pool = await client.create_work_pool(
            work_pool=WorkPoolCreate(
                name="my-process-pool",
                type="process",
                base_job_template=base_job_template,
                is_paused=False,
            )
        )
        print(f"Created work pool with custom defaults: {work_pool.name}")

asyncio.run(main())

Notes - Authenticate your script to your Prefect API (Cloud or Server) via

PREFECT_API_URL

and

PREFECT_API_KEY

(Cloud) or run against a local Server. - Deployment-level

job_variables

will override these pool defaults; runtime job variables override both. Helpful docs - Work pools overview: Work Pools &amp; Workers - Client and work pool API (CLI ref for context): prefect work-pool CLI

Short answer: there isn’t a supported public Python API in Prefect 2.20.6 to “load and deploy from prefect.yaml.” The recommended approach is to call the CLI from your Python script. If you don’t want to use the YAML, you can instead create deployments in code with

flow.deploy()

. Option A — call the CLI from Python (recommended) - Create a specific deployment (non-interactive):

import subprocess

subprocess.run(
    [
        "prefect", "deploy",
        "--prefect-file", "path/to/prefect.yaml",
        "-n", "my-flow/my-deployment"   # or just "my-deployment"
    ],
    check=True
)

- Deploy multiple via name patterns:

import subprocess

subprocess.run(
    [
        "prefect", "deploy",
        "--prefect-file", "path/to/prefect.yaml",
        "-n", "my-flow/*"               # all deployments for this flow
    ],
    check=True
)

- Deploy all in the file by iterating names from YAML (avoids interactive prompts):

import subprocess, yaml, pathlib

prefect_file = pathlib.Path("path/to/prefect.yaml")
cfg = yaml.safe_load(prefect_file.read_text())

for d in cfg.get("deployments", []):
    # Note: passing just the deployment name is fine if names are unique
    subprocess.run(
        [
            "prefect", "deploy",
            "--prefect-file", str(prefect_file),
            "-n", d["name"]
        ],
        check=True
    )

Notes: - Make sure your environment is authenticated to your Prefect API (

PREFECT_API_URL

and

PREFECT_API_KEY

for Cloud, or run against Server). -

--name/-n

supports exact names and patterns like

my-flow/my-deploy

,

my-deployment-*

,

my-flow/*

. Option B — create deployments in code (bypasses prefect.yaml)

from prefect import flow

@flow
def my_flow():
    ...

if __name__ == "__main__":
    my_flow.deploy(
        name="my-deployment",
        work_pool_name="my-process-pool",
        parameters={"foo": "bar"},
        tags=["dev"]
    )

Why not a direct Python API for prefect.yaml? - In 2.20.6, the functions that parse and execute build/push/pull from prefect.yaml are internal/unstable (prefixed with “_”) and not supported as a public API. The stable path is the CLI or defining deployments directly in Python. Docs - Deploy with prefect.yaml: Guide: Deploy with prefect.yaml - CLI reference (prefect deploy): CLI: prefect deploy

Yes. A deployment becomes “process-based” by attaching it to a work pool of type process. The

.deploy

method will create the deployment and point it at whatever work pool you name. Typical steps 1) Create a Process work pool (once) 2) Call

.deploy(...)

on your flow with

work_pool_name

set to that pool 3) Run a Process worker that polls the pool Example: create the pool (optional if it already exists)

import asyncio
from prefect.client.orchestration import get_client
from prefect.client.schemas.actions import WorkPoolCreate
from prefect.workers.process import ProcessWorker

async def ensure_process_pool(name="my-process-pool"):
    base_job_template = ProcessWorker.get_default_base_job_template()
    async with get_client() as client:
        try:
            await client.create_work_pool(
                work_pool=WorkPoolCreate(
                    name=name,
                    type="process",
                    base_job_template=base_job_template,
                    is_paused=False,
                )
            )
            print(f"Created pool: {name}")
        except Exception as e:
            print(f"Pool may already exist: {e}")

asyncio.run(ensure_process_pool())

Deploy your flow to that pool

from prefect import flow

@flow
def my_flow():
    print("Hello from process worker!")

if __name__ == "__main__":
    my_flow.deploy(
        name="my-process-deployment",
        work_pool_name="my-process-pool",
        # optional process job variables:
        job_variables={
            "stream_output": True,
            "working_dir": "/tmp/prefect-runs",
            "env": {"LOG_LEVEL": "INFO"},
            "labels": {"team": "data-eng"}
        },
        # optional schedule helpers:
        cron="0 9 * * *"  # run daily at 09:00
    )

Then start a process worker to pick up runs from that pool

prefect worker start -p my-process-pool

Notes - The worker type is dictated by the work pool;

.deploy

just points to that pool. - Common process job_variables:

stream_output

,

working_dir

,

env

,

labels

(you can also set these as pool defaults in the pool’s base job template). Docs - Work pools and workers: Work Pools &amp; Workers - Deploying flows (2.x): Deploy with prefect.yaml and .deploy

You’re hitting that error because

.deploy

needs a way to make your flow code available to the worker at runtime. In Prefect 2.20.6, that means one of: - Provide a container image (via the

image

arg to

.deploy

), or - Provide a remote source (via a prefect.yaml with pull steps like git/S3) A Process worker does not use images, but

.deploy

still enforces “image or remote storage” so your code can be reproduced on the worker. You have three workable paths: Option 1 — Use prefect.yaml with a pull step (recommended for Process workers) 1) Define a git (or S3/GCS) pull in prefect.yaml:

pull:
  - prefect.deployments.steps.git_clone:
      repository: <https://github.com/your-org/your-repo.git>
      branch: main

deployments:
  - name: pro-process-deploy
    entrypoint: flows/my_flow.py:my_flow
    work_pool:
      name: pro_process

2) Register the deployment with:

prefect deploy -n pro-process-deploy --prefect-file prefect.yaml

Docs: Deploy with prefect.yaml Option 2 — Pass an image to .deploy (unblocks the error, even if Process won’t use the image)

from prefect import flow

@flow
def my_flow():
    print("hi")

if __name__ == "__main__":
    my_flow.deploy(
        name="pro-process-deploy",
        work_pool_name="pro_process",
        image="my-registry/my-image:latest",  # provides a code artifact
        build=False,  # set True to build if you want
        push=False,   # set True to push if you build
        job_variables={"stream_output": True},
    )

Note: The Process worker will ignore the image at run time, but this satisfies `.deploy`’s packaging requirement. Option 3 — Create the deployment via the client (bypass

.deploy

check) This works if the worker can see your code at the same path (e.g., same git checkout on the worker host).

import asyncio
from prefect.client.orchestration import get_client

async def create_deployment():
    async with get_client() as client:
        flow_id = await client.create_flow_from_name("my_flow")
        dep = await client.create_deployment(
            flow_id=flow_id,
            name="pro-process-deploy",
            work_pool_name="pro_process",
            entrypoint="flows/my_flow.py:my_flow",
            tags=["process"],
            parameters={}
        )
        print(f"Created deployment: {dep.name}")

asyncio.run(create_deployment())

Caveat: with this approach, you are responsible for ensuring the worker host has the repo/code at

flows/my_flow.py

. Which should you choose? - If your workers run on separate machines or you want a clean, reproducible setup: use prefect.yaml with a pull step (Option 1). - If you just want to unblock

.deploy

quickly: add

image=...

(Option 2). - If your worker runs on the same machine/path as your code: create the deployment with the client (Option 3). Related docs - Work pools and workers: Work Pools &amp; Workers - Deploy with prefect.yaml and pull steps: Deploy with prefect.yaml

Great question—and you’re very close already. In Prefect 2.20.x, the KubernetesWorker uses a “base job template” that contains a full Kubernetes Job manifest with Jinja placeholders. To migrate a Prefect 1.x KubernetesRun custom-job.yaml to Prefect 2.20’s KubernetesWorker, you’ll: 1) Start from the default base job template for the Kubernetes worker 2) Drop your customizations into the Job manifest section 3) Ensure you keep Prefect’s Jinja placeholders intact so the worker can inject runtime values How to fetch the default template (for reference) - CLI:

prefect work-pool get-default-base-job-template --type kubernetes > base_job_template.json

- Python:

from prefect.workers.kubernetes import KubernetesWorker
tmpl = KubernetesWorker.get_default_base_job_template()

What the base job template looks like - It is a JSON object with two keys: - job_configuration: contains the full Job manifest plus some worker config - variables: JSON Schema describing the variables you can pass (and their defaults) that fill into the manifest/template Important: the Job spec lives under job_configuration.job_manifest. Prefect injects variables using Jinja placeholders such as: - metadata.labels: "{{ labels }}" - metadata.namespace: "{{ namespace }}" - metadata.generateName: "{{ name }}-" - spec.ttlSecondsAfterFinished: "{{ finished_job_ttl }}" - spec.template.spec.serviceAccountName: "{{ service_account_name }}" - container image fields: "{{ image }}" and "{{ image_pull_policy }}" - container args: "{{ command }}" (note: injected into args, not command) - container env: "{{ env }}" Minimal conversion pattern Take your 1.x custom job YAML and place it under job_configuration.job_manifest in the 2.x base job template, then replace the fields you want to be dynamic with Prefect’s placeholders. Example (showing just the relevant JSON portions)

{
  "job_configuration": {
    "namespace": "{{ namespace }}",
    "job_manifest": {
      "apiVersion": "batch/v1",
      "kind": "Job",
      "metadata": {
        "labels": "{{ labels }}",
        "namespace": "{{ namespace }}",
        "generateName": "{{ name }}-"
      },
      "spec": {
        "backoffLimit": 0,
        "ttlSecondsAfterFinished": "{{ finished_job_ttl }}",
        "template": {
          "spec": {
            "restartPolicy": "Never",
            "serviceAccountName": "{{ service_account_name }}",
            "containers": [
              {
                "name": "prefect-job",
                "image": "{{ image }}",
                "imagePullPolicy": "{{ image_pull_policy }}",
                "env": "{{ env }}",
                "args": "{{ command }}"
              }
            ]
          }
        }
      }
    },
    "cluster_config": "{{ cluster_config }}",
    "job_watch_timeout_seconds": "{{ job_watch_timeout_seconds }}",
    "pod_watch_timeout_seconds": "{{ pod_watch_timeout_seconds }}",
    "stream_output": "{{ stream_output }}"
  },
  "variables": {
    "type": "object",
    "properties": {
      "namespace": {
        "type": "string",
        "default": "default"
      },
      "image": {
        "type": [
          "string",
          "null"
        ],
        "default": null
      },
      "service_account_name": {
        "type": [
          "string",
          "null"
        ],
        "default": null
      },
      "image_pull_policy": {
        "type": "string",
        "enum": [
          "IfNotPresent",
          "Always",
          "Never"
        ],
        "default": "IfNotPresent"
      },
      "finished_job_ttl": {
        "type": [
          "integer",
          "null"
        ],
        "default": null
      },
      "job_watch_timeout_seconds": {
        "type": [
          "integer",
          "null"
        ],
        "default": null
      },
      "pod_watch_timeout_seconds": {
        "type": "integer",
        "default": 60
      },
      "stream_output": {
        "type": "boolean",
        "default": true
      },
      "cluster_config": {
        "type": [
          "object",
          "null"
        ],
        "default": null
      },
      "env": {
        "type": "object",
        "default": {}
      },
      "labels": {
        "type": "object",
        "default": {}
      },
      "command": {
        "type": [
          "array",
          "null"
        ],
        "default": null
      },
      "name": {
        "type": [
          "string",
          "null"
        ],
        "default": null
      }
    }
  }
}

Notes and gotchas when converting

- args vs command: Prefect injects the runtime command as args ("{{ command }}") in the container spec. If your 1.x manifest used

command:

and

args:

, prefer keeping Prefect’s args placeholder and let the worker supply it. - env format: In your template, use the placeholder "{{ env }}". Prefect will convert a dict of env vars into the Kubernetes list-of-dicts format at runtime. - labels: Use "{{ labels }}"; Prefect will normalize label keys/values. - generateName: Prefer using metadata.generateName with "{{ name }}-" so Prefect can ensure unique job names per run. - Keep placeholders intact: Don’t hardcode values that Prefect needs to fill (image, command, env, labels, namespace, service_account_name, image_pull_policy, finished_job_ttl). How to apply your custom template in a pool - Create or update a Kubernetes work pool with your adapted base template (UI → Work Pools → your pool → Advanced). - Then, in your deployments, pass job_variables to set values for image, namespace, env, etc. Example:

work_pool:
  name: my-k8s-pool
  job_variables:
    image: "my-registry/my-image:latest"
    namespace: "prefect"
    service_account_name: "prefect-worker"
    image_pull_policy: "IfNotPresent"
    finished_job_ttl: 3600
    env:
      LOG_LEVEL: INFO

Helpful commands - Show the default template to use as a base:

prefect work-pool get-default-base-job-template --type kubernetes

Docs - Work pools and workers: Work Pools &amp; Workers - Deploy with work pools (Kubernetes): Deploy with prefect.yaml - Overriding job variables: Overriding Work Pool Job Variables If you paste a snippet of your 1.x job YAML, I can help translate it 1:1 into a Prefect 2.20 base job template with the right placeholders.

You don’t pass a file path to the KubernetesWorker directly. Instead, you load your custom base job template file in Python, parse it into a dict, and supply it as the base_job_template when you create or update the Kubernetes work pool. Create a new Kubernetes work pool with your template

import asyncio, json
from prefect.client.orchestration import get_client
from prefect.client.schemas.actions import WorkPoolCreate

async def main():
    # Load your adapted base job template (JSON or YAML parsed to a dict)
    with open("my_k8s_base_job_template.json") as f:
        template = json.load(f)

    async with get_client() as client:
        pool = await client.create_work_pool(
            work_pool=WorkPoolCreate(
                name="my-k8s-pool",
                type="kubernetes",
                base_job_template=template,   # ← pass the dict here
                is_paused=False,
                description="K8s pool with custom base job template"
            )
        )
        print(f"Created pool: {pool.name}")

asyncio.run(main())

Update an existing pool’s template

import asyncio, json
from prefect.client.orchestration import get_client
from prefect.client.schemas.actions import WorkPoolUpdate

async def main():
    with open("my_k8s_base_job_template.json") as f:
        template = json.load(f)

    async with get_client() as client:
        await client.update_work_pool(
            work_pool_name="my-k8s-pool",
            work_pool=WorkPoolUpdate(
                base_job_template=template   # ← update with your dict
            )
        )
        print("Pool template updated")

asyncio.run(main())

YAML template support If your file is YAML:

import yaml
with open("my_k8s_base_job_template.yaml") as f:
    template = yaml.safe_load(f)

What must be in the template - Top-level keys:

job_configuration

and

variables

- The full Kubernetes Job manifest must be at

job_configuration.job_manifest

- Keep Prefect’s placeholders in the manifest so deployments can override them: -

{{ namespace }}

,

{{ labels }}

,

{{ name }}

,

{{ finished_job_ttl }}

-

{{ service_account_name }}

,

{{ image }}

,

{{ image_pull_policy }}

-

{{ env }}

,

{{ command }}

(in containers[].args) - The

variables

section is a JSON Schema object defining those placeholders and defaults Tip: To see the stock template you’re replacing:

prefect work-pool get-default-base-job-template --type kubernetes > default_template.json

Once set on the pool, any deployment targeting that pool can override values via

work_pool.job_variables

(e.g., image, namespace, env), and your custom manifest will be used by the Kubernetes worker at runtime.

Short answer: not exactly. In Prefect 2.20.6, priorities within a single work pool must be unique. If you try to give two queues the same priority, Prefect will rebalance to keep priorities unique and contiguous (lower number = higher priority). What happens in practice - If you omit priority, Prefect auto-assigns the first available slot (e.g., 1, then 2, then 3; or it fills gaps like 1,2,4 → assigns 3). - If you explicitly set a duplicate (e.g., create two queues both with priority=5), Prefect will adjust priorities so each queue ends up with a unique value. You cannot end up with two queues at the same priority in one pool. How to set priority - Python:

import asyncio
from prefect.client.orchestration import get_client

async def main():
    async with get_client() as client:
        await client.create_work_queue(
            name="high-pri",
            work_pool_name="my-pool",
            priority=1
        )
        await client.create_work_queue(
            name="med-pri",
            work_pool_name="my-pool",
            priority=2
        )

asyncio.run(main())

- CLI:

prefect work-queue create high-pri --pool my-pool --priority 1
prefect work-queue create med-pri  --pool my-pool --priority 2

Key takeaways - Priorities are unique per work pool; two queues cannot share the same priority. - Lower numbers mean higher priority (1 is highest). - Prefect will auto-assign or rebalance to maintain unique, sequential priorities.

Great question! Here’s a concise rundown of the biggest changes to flows and tasks from Prefect 1 → Prefect 2 (2.20.x), with side‑by‑side examples. High‑level changes - Definitions - Prefect 1: flows built with a context manager; tasks can be functions or Task subclasses. - Prefect 2: flows and tasks are just Python functions decorated with @flow and @task. No class inheritance needed. - Parameters - Prefect 1:

Parameter

objects inside the flow context. - Prefect 2: normal function arguments (with optional type hints). The parameter schema is inferred from the signature. - Mapping/concurrency - Prefect 1:

.map()

for parallelism. - Prefect 2:

.submit()

returns Prefect futures; use normal Python loops/comprehensions. Concurrency is controlled by the flow’s task runner. - Control flow and dependencies - Prefect 1: DSL (edges, triggers, signals) + context-based relationships. - Prefect 2: native Python control flow (if/for/try) and dataflow dependencies through passing futures/values. No triggers/signals; raise exceptions to fail. - Execution - Prefect 1:

flow.run()

for local; agents for orchestration. - Prefect 2: call the flow like a function for local; orchestration via deployments (workers/work pools). - Scheduling - Prefect 1: schedules attached to flows. - Prefect 2: schedules attached to deployments (decoupled from code). - Results and caching - Prefect 1: Result objects, checkpointing, targets. - Prefect 2: simple result persistence/caching via

@task(persist_result=..., cache_key_fn=..., cache_expiration=...)

. - Logging and context - Prefect 1:

prefect.context

and task loggers. - Prefect 2:

get_run_logger()

for logs; runtime info via

prefect.runtime

or

prefect.context.get_run_context()

. - Subflows - Prefect 1: calling flows within tasks discouraged/complex. - Prefect 2: subflows are first‑class; call one

@flow

from another like a normal function. Side‑by‑side examples Prefect 1.x

from datetime import timedelta
from prefect import task, Flow, Parameter

@task(max_retries=2, retry_delay=timedelta(seconds=10))
def add(x, y):
    return x + y

with Flow("my-flow") as flow:
    n = Parameter("n", default=1)
    # Parallel map
    results = add.map(range(5), unmapped(n))

# Local execution
state = flow.run()

Prefect 2.x (2.20)

from prefect import flow, task, get_run_logger

@task(retries=2, retry_delay_seconds=10, persist_result=True)
def add(x, y):
    logger = get_run_logger()
    out = x + y
    <http://logger.info|logger.info>(f"add: {out}")
    return out

@flow
def my_flow(n: int = 1):
    # Parallel submission
    futures = [add.submit(i, n) for i in range(5)]
    return [f.result() for f in futures]

if __name__ == "__main__":
    my_flow()  # local run

Other notable diffs - Retries - 1.x:

@task(max_retries=..., retry_delay=timedelta(...))

- 2.x:

@task(retries=..., retry_delay_seconds=...)

(flows can have retries too) - Task runners (parallelism) - 1.x: executors (Local/Dask). - 2.x: task runners on the flow (e.g., ConcurrentTaskRunner, DaskTaskRunner). Example:

from prefect import flow
    from prefect.task_runners import ConcurrentTaskRunner

    @flow(task_runner=ConcurrentTaskRunner())
    def my_flow():
        ...

- Packaging and code location - 1.x: Flow.storage (Local/GitHub/S3) set on the Flow object. - 2.x: code packaging handled by deployments (e.g., git_clone pull step) or container images via build steps in prefect.yaml. Docs to dive deeper - Flows (2.x): Flows in Prefect 2 - Tasks (2.x): Tasks in Prefect 2 - Deployments and packaging: Deploy with prefect.yaml - Logging and runtime info: Logging and Runtime context

If you share a small Prefect 1 flow/task snippet you’re migrating, I can translate it exactly to Prefect 2 style.

Yes, you can stack other decorators with Prefect’s @task, but the order matters and there are a few caveats. Recommended pattern - Put @task as the outermost decorator (top line). This ensures Prefect wraps a plain function, not another object. - Make sure your additional decorators use functools.wraps to preserve the original function’s name and signature. Prefect infers task metadata and parameter schema from the function signature. Example (good)

from functools import wraps
from prefect import task

def log_calls(fn):
    @wraps(fn)
    def wrapper(*args, **kwargs):
        print(f"Calling {fn.__name__}")
        return fn(*args, **kwargs)
    return wrapper

@task
@log_calls
def add(x: int, y: int) -> int:
    return x + y

Example (risky order)

@log_calls
@task
def add(x, y):
    return x + y

Here,

log_calls

wraps a Prefect Task object instead of a plain function. It may still work, but it’s brittle: your decorator may not expect a Task object, you can lose the original signature/metadata, and debugging is harder. Tips and caveats - Signature preservation: If your decorator doesn’t use functools.wraps (or otherwise alters the signature), Prefect may infer the wrong parameter schema. If needed, set task options explicitly:

@task(name="my_task")

. - Async vs sync: Prefect supports both; wrapping a sync function with an async decorator (or vice versa) is fine as long as it remains a callable and your decorator handles the await correctly. - Don’t replace the function with a non-callable: Decorators that turn the function into a property/descriptor will break task creation. - Side effects and retries: Prefer Prefect’s built-in task controls (

retries

,

retry_delay_seconds

,

timeout_seconds

) over external retry/time-limit decorators that can conflict with orchestration semantics. - Caching decorators: e.g.,

functools.lru_cache

can be stacked, but remember it only caches within the same process and won’t persist across workers; Prefect’s `persist_result`/`cache_key_fn` are better for orchestrated caching. If you share the specific decorator you want to combine, I can suggest the safest ordering and any adjustments.

Short answer: there’s no direct equivalent to Prefect 1’s triggers (like

any_failed

) in Prefect 2.20. Triggers were removed. In 2.x you handle this with explicit state inspection or callbacks. Common patterns you can use: - Fan-out, inspect states, then conditionally run a follow-up task (replacement for any_failed)

from prefect import flow, task, get_run_logger

@task
def work(i: int):
    if i == 2:
        raise ValueError("boom")
    return i

@task
def notify(failed_ids: list[int]):
    logger = get_run_logger()
    logger.warning(f"Failures detected for: {failed_ids}")

@flow
def main():
    futs = [work.submit(i) for i in range(5)]
    failed_ids = [i for i, f in enumerate(futs) if f.wait().is_failed()]
    if failed_ids:
        notify(failed_ids)

- Task-level failure callbacks (runs when that task enters FAILED)

from prefect import flow, task, get_run_logger

def on_fail(task, task_run, state):
    logger = get_run_logger()
    logger.error(f"{task.name} failed: {state.message}")

@task(on_failure=[on_fail])
def risky():
    raise RuntimeError("nope")

@flow
def main():
    risky()

- Pull states directly without raising, then branch

state = some_task(return_state=True)
if state.is_failed():
    # do something when it failed
    ...

Tips - Use

future.wait()

+

state.is_failed()

for parallel patterns. - To read exceptions without raising:

state.result(raise_on_failure=False)

or

prefect.states.get_state_exception(state)

. - You can also attach a flow-level

on_failure=[...]

if you want a single handler when the whole flow fails. Docs - States - Tasks - Migration notes (1.x → 2.x): Migration guide

Short answer: there’s no direct equivalent to Prefect 1’s

set_reference_tasks

in Prefect 2.20. In 2.x, a flow’s final state is determined by what happens inside the flow function: if an exception is raised (including when you call

.result()

on a failed task), the flow fails; otherwise it succeeds. There’s no special “reference task” concept. How to replicate “reference task” behavior in 2.x - Only let the tasks you “care about” influence the flow state. Concretely: - Await or call

.result()

on the “reference” tasks so their failures will fail the flow - For non-reference tasks, either don’t call

.result()

at all, or call it with

raise_on_failure=False

, or wrap in try/except to swallow/log their errors Example: fail the flow only if any “reference” tasks fail

from prefect import flow, task, get_run_logger

@task
def important(i):
    if i == 2:
        raise ValueError("important failed")
    return f"ok-{i}"

@task
def ancillary(i):
    # This may fail, but should not fail the flow
    if i == 3:
        raise RuntimeError("ancillary failed")
    return f"anc-{i}"

@task
def log_failure(msg: str):
    logger = get_run_logger()
    logger.warning(msg)

@flow
def my_flow():
    # Submit both kinds of tasks
    imp_futs = [important.submit(i) for i in range(5)]
    anc_futs = [ancillary.submit(i) for i in range(5)]

    # “Reference” tasks: fail the flow if any failed
    failed_imps = [f for f in imp_futs if f.wait().is_failed()]
    if failed_imps:
        raise RuntimeError(f"{len(failed_imps)} important tasks failed")

    # Non-reference tasks: inspect/log but don’t fail the flow
    for f in anc_futs:
        st = f.wait()
        if st.is_failed():
            log_failure.submit(f"Ancillary failed: {st.message}")

    return "done"

Variants - If you need the result but don’t want failures to raise: -

val_or_exc = future.result(raise_on_failure=False)

and check

isinstance(val_or_exc, Exception)

- If you want the flow to succeed even if a “reference” task fails, catch it: -

try: x = future.result(); except Exception: ...

(don’t re-raise) Useful docs - Flows: Flows in Prefect 2 - Tasks: Tasks in Prefect 2 - States: States If you share your 1.x pattern (which tasks you marked as references), I can translate it to a clean 2.x control-flow pattern.

Short answer - on_failure: runs when the flow ends in a FAILED terminal state (e.g., an uncaught exception in your code after retries are exhausted). - on_crashed: runs when the flow ends in a CRASHED terminal state (e.g., the Python process/container is killed, OOM, segfault, or otherwise can’t report back cleanly). They are mutually exclusive for a given run. If you want something to run for any terminal outcome, use on_completion. Key nuances - Failures happen “inside” your code path; the engine can still execute your failure hook within the run’s context. - Crashes happen “outside” your code path; the process that was running your flow died. The crash callback is triggered during orchestration state handling, not inside the dead process. Don’t depend on local process resources in on_crashed; use it to notify or record events. Minimal example

from prefect import flow, get_run_logger

def notify_failure(flow, flow_run, state):
    logger = get_run_logger()
    logger.error(f"[FAILURE] {flow.name} run {flow_run.id}: {state.message}")

def notify_crash(flow, flow_run, state):
    logger = get_run_logger()
    logger.critical(f"[CRASH] {flow.name} run {flow_run.id}")

@flow(
    on_failure=[notify_failure],
    on_crashed=[notify_crash],
)
def demo_flow(should_crash: bool = False):
    if should_crash:
        # Simulate a hard crash: kills the process abruptly
        import os
        os._exit(1)   # Note: your flow code after this will not run
    # Simulate a “normal” failure (caught by on_failure)
    raise RuntimeError("Boom!")

# Examples:
# demo_flow()            -> ends as FAILED; on_failure runs
# demo_flow(True)        -> ends as CRASHED; on_crashed runs (not in the killed process)

Tips - Retries: on_failure runs only after the final attempt fails. Intermediate failed attempts transition to AwaitingRetry instead. - If you want to react to any terminal outcome (Completed/Failed/Cancelled/Crashed), use on_completion as a catch‑all. - For production notifications on crashes, prefer Prefect Cloud Automations/Webhooks so the alert does not depend on the crashed process. Docs - Flow callbacks and states: Flows and States - Automations (Cloud) for alerts: Automations

@Marvin When using

on_cancellation

, can I assign a state to the flow run inside the handler method?

Short answer: No. The on_cancellation handler is a side-effect hook. It cannot change the flow run’s final state by returning a value or “assigning” a new state. Use it for cleanup/notifications only. Details - Flow/Task callbacks in Prefect 2.20 (on_completion, on_failure, on_cancellation, on_crashed, etc.) are invoked when the run transitions to that state. They receive (flow, flow_run, state) and run in the flow’s process. - Returning a different state from the handler, or raising in the handler, does not override the already-determined state. - Even if you try to call the orchestration API to propose a different state (e.g., using the client inside the handler), transitions from a terminal state like Cancelled will generally be rejected by orchestration rules. Practically, the engine is shutting the run down at this point. What to do instead - If you want custom “final outcome” logic, implement it before the run is cancelled (e.g., catch your own condition and exit cleanly so the flow completes rather than cancels). - Use on_cancellation to: - log/notify - create artifacts or events - enqueue compensating work (e.g., start another flow run) without attempting to change the current run’s state Example

from prefect import flow, get_run_logger

def on_cancel(flow, flow_run, state):
    logger = get_run_logger()
    logger.warning(f"Flow {flow_run.name} was cancelled: {state.message}")
    # OK: notify, emit artifacts, enqueue follow-up work
    # NOT OK: trying to reassign the flow state here

@flow(on_cancellation=[on_cancel])
def my_flow(should_cancel: bool = False):
    if should_cancel:
        # Simulate cooperative cancellation
        import asyncio
        # e.g., an external cancel will set the run to Cancelling/Cancelled;
        # just ensure your code can exit gracefully
        asyncio.sleep(0.1)
    return "done"

If you share what you were hoping to accomplish by “assigning a state,” I can suggest a pattern that fits (e.g., finishing as Completed with a warning, creating a compensating run, or annotating results). Links: States overview States, Flows and callbacks Flows.

Short answer: the on_cancellation hook cannot change the flow’s state. It runs after the flow has already entered Cancelling, and its return value is ignored. Use it for cleanup/notifications only. If your goal is “run some handler, then have the flow finish Cancelled,” you have two supported patterns: A) Cancel from inside the flow (graceful, self‑directed) - Implement your handler logic directly in the flow (or a task it calls), then return a Cancelled state as the flow’s final value. - Returning a Prefect state from a flow determines the final flow state. Example:

from prefect import flow
from prefect.states import Cancelled

def my_cleanup():
    # your handler logic here
    pass

@flow
def my_flow(should_cancel: bool = False):
    if should_cancel:
        my_cleanup()                 # run your handler work
        return Cancelled(message="User-requested cancellation after cleanup")
    # ... normal work ...
    return 42

# Verify
st = my_flow(should_cancel=True, return_state=True)
assert st.is_cancelled()

B) Cancel from outside the flow (orchestrator‑directed) - If the cancellation is initiated externally (UI/CLI/automation/API), use the on_cancellation hook just for side‑effects. - The system will transition the run from Cancelling → Cancelled after your hook runs; you cannot override that sequence in the hook. External cancel example:

import asyncio
from prefect.client.orchestration import get_client
from prefect.states import Cancelling

async def cancel_run(flow_run_id):
    async with get_client() as client:
        await client.set_flow_run_state(flow_run_id, Cancelling(message="Ops cancel"))

# elsewhere: asyncio.run(cancel_run(<flow_run_id>))

Important notes - on_cancellation fires when the run enters the non‑terminal Cancelling state; it cannot reassign state. - If you want to guarantee “do X then end Cancelled” from inside your code, do X and return Cancelled() from the flow. - Do not raise CancelledRun to try to cancel; that exception is used when reading results from an already‑cancelled run. Related docs - States: States - Flows and callbacks: Flows

Yes, you can. The

prefect_test_harness

will use whatever database your Prefect API is configured to use. By default it points to a temp SQLite DB, but you can override it to PostgreSQL by setting

PREFECT_API_DATABASE_CONNECTION_URL

before entering the harness. Quick example (Python)

from prefect.testing.utilities import prefect_test_harness
from prefect.settings import temporary_settings, PREFECT_API_DATABASE_CONNECTION_URL

POSTGRES_URL = "<postgresql+asyncpg://user:pass@localhost/prefect_test>"

with temporary_settings({PREFECT_API_DATABASE_CONNECTION_URL: POSTGRES_URL}):
    with prefect_test_harness(server_startup_timeout=60):
        # Your tests run against the Postgres-backed ephemeral API server
        ...

Key points and caveats - SQLite is isolated automatically: the harness creates and tears down a fresh temp DB file per run. - PostgreSQL is not isolated automatically: you must manage DB lifecycle/cleanup yourself (e.g., create a dedicated test database, truncate tables between tests, or create per-worker DBs if running pytest-xdist). - The harness spins up an ephemeral API server in a subprocess and runs migrations on startup; your PostgreSQL server must be running and reachable. - Use a proper async PG URL:

<postgresql+asyncpg://user:pass@host:port/dbname>

Tips for larger suites - Session-scoped fixture in conftest.py that sets the Postgres URL via

temporary_settings(...)

and wraps tests in

prefect_test_harness(...)

. - For parallel tests (xdist), create one DB per worker (e.g., suffix by worker_id) and drop them afterward. - If you need table cleanup between tests, add an autouse fixture to truncate tables. When to use Postgres vs SQLite - Use SQLite for fast, isolated unit tests. - Use PostgreSQL for integration tests that need to mirror production behavior. If you share how you run your tests (pytest, xdist, CI), I can provide a ready-to-use conftest.py that sets up per-worker PostgreSQL databases and handles cleanup.

Sorry, I encountered an error while processing your request. Please try again.

Short answer: the cleanest solution is to have the flow return what you need. If that’s not possible, you can still retrieve task results after the run by querying task runs from the API and deserializing their states. Two workable approaches: A) Best practice: return values from the flow - Make the flow pass through what your test needs (e.g., a dict of task results). This is simplest and works in all execution modes. B) If you can’t change the flow: query task runs after execution - Run the flow with return_state=True to get the flow_run_id - Use the Prefect client to list task runs for that flow run - For each task run, deserialize the result from its state - Important: this requires running against a Prefect API (e.g., with prefect_test_harness or Cloud/Server). If the tasks ran on remote workers, ensure persist_result=True and result storage are configured. Example pytest-friendly snippet

import asyncio
from prefect.client.orchestration import get_client
from prefect.states import get_state_result

def run_flow_and_collect_task_results(flow_fn, *, task_name_filter=None):
    # Run the flow and capture the flow state (includes flow_run_id)
    flow_state = flow_fn(return_state=True)
    flow_run_id = flow_state.state_details.flow_run_id
    if not flow_run_id:
        raise RuntimeError(
            "No flow_run_id available. Make sure you're running against a Prefect API "
            "(e.g., use prefect_test_harness or Prefect Cloud/Server)."
        )

    async def _collect():
        async with get_client() as client:
            task_runs = await client.read_task_runs(flow_run_id=flow_run_id)
            results = {}
            for tr in task_runs:
                if task_name_filter and tr.name != task_name_filter:
                    continue
                # Deserialize the Python value from the task's final state
                results[tr.name, str(tr.id)] = get_state_result(tr.state)
            return results

    return asyncio.run(_collect())

Usage in a test fixture

@pytest.fixture
def task_results(prefect_api_harness):  # e.g., your prefect_test_harness wrapper
    # Assuming my_flow() contains tasks that return values but the flow returns nothing
    from my_project.flows import my_flow

    # Optionally filter to a specific task by name
    results = run_flow_and_collect_task_results(my_flow, task_name_filter="compute")
    return results

def test_something(task_results):
    # task_results is a dict keyed by (task_name, task_run_id) -> value
    assert any(v == 123 for v in task_results.values())

Notes and caveats - Without an API backend (pure in-process run), you can’t query task runs; either return values from the flow or emit artifacts and read them back. - For remote workers, set persist_result=True on tasks and configure result_storage so results can be retrieved by the test process. - You can also use artifacts inside tasks (e.g., create_table/create_markdown) and fetch them via the artifacts API filtered by flow_run_id if you prefer not to persist task results. If you share a small example of your flow/tasks and how you run tests (local vs prefect_test_harness vs Cloud), I can tailor the snippet to your setup.

Yes. In Prefect 2.20 (2.x), you can call the REST API directly without the Prefect client. Key point: server-side filtering by flow run parameters is not supported in 2.x. The

/flow_runs/filter

endpoint can filter by many fields (id, name, tags, deployment_id, state, time ranges, etc.), but not by the

parameters

JSON. To find runs by a parameter value, fetch a narrowed set of runs, then filter client-side on the

parameters

field. Examples - Base URLs: - Prefect Server (self-hosted):

<http://localhost:4200/api>

- Prefect Cloud 2:

<https://api.prefect.cloud/api/accounts/<ACCOUNT_ID>/workspaces/<WORKSPACE_ID>>

- cURL (self-hosted) with client-side parameter filtering via jq:

API_URL="<http://localhost:4200/api>"

curl -s -X POST "$API_URL/flow_runs/filter" \
  -H "Content-Type: application/json" \
  -d '{
    "flow_runs": {
      "deployment_id": {"any_": ["YOUR-DEPLOYMENT-ID"]},
      "start_time": {"after_": "2024-01-01T00:00:00Z"}
    },
    "limit": 200,
    "offset": 0
  }' \
| jq '.[] | select(.parameters.my_param == "target_value") | {id, name, parameters}'

- cURL (Cloud 2) with client-side parameter filtering:

ACCOUNT_ID="..."
WORKSPACE_ID="..."
API_KEY="prefect-cloud-api-key"

curl -s -X POST "<https://api.prefect.cloud/api/accounts/$ACCOUNT_ID/workspaces/$WORKSPACE_ID/flow_runs/filter>" \
  -H "Content-Type: application/json" \
  -H "Authorization: Bearer $API_KEY" \
  -d '{
    "flow_runs": {
      "tags": {"all_": ["important"]},
      "state": {"type": {"any_": ["COMPLETED","FAILED"]}}
    },
    "limit": 200
  }' \
| jq '.[] | select(.parameters.my_param == "target_value") | {id, name, parameters}'

- Python requests with client-side filtering:

import requests

api_url = "<http://localhost:4200/api>"  # or Cloud base URL
headers = {"Content-Type": "application/json"}  # add Authorization for Cloud

payload = {
    "flow_runs": {
        "deployment_id": {"any_": ["YOUR-DEPLOYMENT-ID"]},
        "start_time": {"after_": "2024-01-01T00:00:00Z"}
    },
    "limit": 200,
    "offset": 0
}

r = <http://requests.post|requests.post>(f"{api_url}/flow_runs/filter", json=payload, headers=headers)
r.raise_for_status()
runs = r.json()

filtered = [fr for fr in runs if fr.get("parameters", {}).get("my_param") == "target_value"]
for fr in filtered:
    print(fr["id"], fr["name"], fr.get("parameters"))

If filtering-by-parameter is critical, a common pattern is to: - copy the parameter value into a tag when creating the run (so you can filter server-side by tags), or - include it in the flow run name or idempotency key for easier retrieval. If you share whether you’re on Cloud vs. self-hosted and the parameter name/value you’re filtering on, I can tailor the example further.

Short answer: there’s no

prefect backend

command in Prefect 2.x. In 2.20, you switch between Cloud and a local Server by setting your profile/config. Equivalent of

prefect backend server

in Prefect 2.20: 1) Start the local Server

prefect server start

2) Point your current profile at the local Server API

prefect config set PREFECT_API_URL=<http://127.0.0.1:4200/api>

Optional (nice workflow using profiles):

prefect profile create local-server
prefect profile use local-server
prefect config set PREFECT_API_URL=<http://127.0.0.1:4200/api>
prefect server start

Related commands (verified): - View current config:

prefect config view

- Manage profiles:

prefect profile ls | create | use | inspect

- Stop server:

prefect server stop

- Switch to Cloud:

prefect cloud login --key <API_KEY> --workspace <account/workspace>

Great question. For Prefect 2.20, you can get the full list of settings (and their environment variable names) here: - Settings guide with explanations and how to configure via env vars and profiles: Prefect 2.20 Settings guide - Complete API reference of all settings (includes names, types, defaults, and descriptions): prefect.settings API reference From the CLI, you can list every setting and its current/default value (useful for discovery), but note it does not include the long-form descriptions:

prefect config view --show-defaults --show-sources

A few handy tips: - Every setting maps to an environment variable (typically uppercase with the

PREFECT_

prefix), e.g.

PREFECT_API_URL

,

PREFECT_HOME

,

PREFECT_LOGGING_LEVEL

. - Use these to change values in your active profile:

prefect config set PREFECT_API_URL=<http://127.0.0.1:4200/api>
prefect config unset PREFECT_API_URL

If you’re targeting a specific area (API, server, logging, database, workers), tell me which and I can point you directly to the relevant settings.

Short answer: No. In Prefect 2.20, a single worker can only poll a single work pool. Workers can, however, poll multiple work queues within that one pool. What you can do: - Poll multiple queues in the same pool:

prefect worker start -p my-pool -q queue1 -q queue2 -q queue3

- If you need to cover multiple pools, run separate worker processes (one per pool):

prefect worker start -p pool-a --name worker-a
prefect worker start -p pool-b --name worker-b

References: - Work pools and workers (2.x concepts) - Worker CLI reference

You're hitting a known snag in Prefect 2.20 when running the server as a non-root user: on startup, the server tries to create/copy the UI build into site-packages at /usr/local/lib/python3.10/site-packages/prefect/server/ui_build which isn’t writable in the official Kubernetes image (it runs as non-root). That’s why you see PermissionError. Two solid ways to fix it: 1) Point the UI static directory to a writable path (recommended) - Create a writable mount inside the container and tell Prefect to use it for the UI static files. - Example patch to your pod spec:

env:
  - name: PREFECT_UI_STATIC_DIRECTORY
    value: /var/lib/prefect/ui_static
# Optional but recommended in k8s so the service can reach it
# (otherwise the default bind is 127.0.0.1)
args:
  - prefect
  - server
  - start
  - --host
  - 0.0.0.0
  - --port
  - "4200"
volumeMounts:
  - name: ui-static
    mountPath: /var/lib/prefect/ui_static
volumes:
  - name: ui-static
    emptyDir: {}

2) Disable the UI entirely (if you only need the API) - Set this env var to skip the UI startup:

env:
  - name: PREFECT_UI_ENABLED
    value: "false"

- You’ll still get the API on the specified port. Other notes: - If you stick with the UI, ensure

PREFECT_UI_API_URL

points to the API route your UI will call (e.g., http//&lt;service host&gt;4200/api). - As a last resort, you could run as root or loosen filesystem permissions on site-packages, but that’s generally not recommended. References: - Hosting Prefect (self-hosted) guide - Official Prefect Server Helm chart (handles these details for you)

Short answer: that flow run is not tied to a deployment, so workers will not pick it up.

create_flow_run(...)

creates an ad‑hoc run that sits in Pending; workers only poll for Scheduled runs created from a deployment in their work pool. In Prefect 3.x: - Workers execute flow runs that are created from a deployment and are in Scheduled. -

create_flow_run(...)

does not attach the run to a deployment, so it stays Pending and is ignored by workers. How to trigger a run the worker will execute Recommended (simple): use

run_deployment

from prefect.deployments import run_deployment

# flow_name/deployment_name
flow_run = run_deployment(
    name="optimize_flow/my-deployment",
    parameters=data,
    timeout=0  # return immediately; omit to wait for completion
)

Lower-level: use the client against a deployment and schedule the run

from prefect.client.orchestration import get_client
from prefect.states import Scheduled
from uuid import UUID

deployment_id = UUID("your-deployment-uuid")

with get_client(sync_client=True) as client:
    flow_run = client.create_flow_run_from_deployment(
        deployment_id=deployment_id,
        parameters=data,
        state=Scheduled()  # important: make it eligible for workers
    )

If you actually want to run the flow right now in the same Python process (no worker), just call it:

optimize_flow(**data)

Checklist if it still won’t start after using a deployment - The deployment uses the same work pool the worker is subscribed to. - The worker is running and healthy (logs show it’s polling that pool). - The deployment is not paused and any concurrency limits are not exhausted. Docs - Run deployments programmatically: How to: Run deployments - Deployments concept: Deployments - Workers concept: Workers If you share how you created the deployment (name and work pool) and how you started the worker, I can help double-check the wiring.

Thanks for clarifying you’re on Prefect 2.20. What you’re seeing is expected:

create_flow_run(...)

creates an ad-hoc flow run that is not tied to a deployment or work queue, so it sits in Pending and agents/workers will not pick it up. Use one of these 2.x approaches instead: Option A: Programmatic trigger (recommended) with run_deployment

from prefect.deployments import run_deployment
import asyncio

async def main():
    flow_run = await run_deployment(
        name="optimize_flow/my-deployment",  # flow_name/deployment_name
        parameters={"your_param": "value"},
        timeout=0  # return immediately; omit to wait for completion
    )
    print(flow_run.id)

asyncio.run(main())

Option B: Low-level client tied to a deployment (ensure Scheduled)

from prefect import get_client
from prefect.states import Scheduled
import asyncio, pendulum
from uuid import UUID

async def main():
    deployment_id = UUID("your-deployment-uuid")
    async with get_client() as client:
        flow_run = await client.create_flow_run_from_deployment(
            deployment_id=deployment_id,
            parameters={"your_param": "value"},
            state=Scheduled(scheduled_time=pendulum.now("UTC"))
        )
        print(flow_run.id)

asyncio.run(main())

CLI equivalent in 2.x

# schedule immediately
prefect deployment run "optimize_flow/my-deployment"

# with parameters
prefect deployment run "optimize_flow/my-deployment" --param your_param=value

# schedule for later
prefect deployment run "optimize_flow/my-deployment" --start-in "5 minutes"

# watch logs
prefect deployment run "optimize_flow/my-deployment" --watch

Why your current approach stays Pending -

create_flow_run()

in 2.20 defaults to Pending, has no deployment_id, and no work_queue_id/name, so agents don’t poll it. - Agents/workers poll work queues for Scheduled runs created from deployments. If it still doesn’t start after switching to a deployment-based trigger, double-check: - The deployment’s work queue matches the agent/worker’s queue/pool. - The agent/worker is running and polling that queue. - The deployment isn’t paused and any concurrency limits aren’t exceeded. Docs (2.x) - Run deployments programmatically: run_deployment API - Client API: Client orchestration API - Deployments: Deployments - Agents: Agents - CLI deployment run: prefect deployment run