Short answer: In Prefect 3.x, parameters defined in prefect.yaml are validated by a Pydantic v2 model that Prefect builds from your flow function’s type hints. To get enum validation, annotate the parameter with your Enum type and ensure the deployment enforces the parameter schema. What to do: 1) Define your enum and annotate the flow parameters - Prefer a str-backed Enum so YAML strings validate cleanly.

from enum import Enum
from pydantic import Field
from prefect import flow

class RunMode(str, Enum):
    ITERATIVE = "iterative"
    BATCH = "batch"

@flow  # validate_parameters=True by default
def my_flow(
    environment: str = "prod",
    mode: RunMode = RunMode.ITERATIVE,   # Enum annotation is key
    window_hours: int = 1,
    dry_run: bool = False,
    sampling_rate: float = Field(1.0, ge=0.0, le=1.0),  # optional: add constraints
):
    ...

2) Set parameters in prefect.yaml and enforce the schema - Make sure to use straight quotes, not curly “smart quotes”.

deployments:
  - name: my-deployment
    entrypoint: path.to.module:my_flow
    parameters:
      environment: prod
      mode: iterative
      window_hours: 1
      dry_run: false
      sampling_rate: 0.1
    enforce_parameter_schema: true

How it behaves: - With the enum annotation, Prefect generates a schema that restricts

mode

to ["iterative", "batch"]. - With

enforce_parameter_schema: true

(recommended), invalid values in the YAML (e.g.,

mode: wrong

) are rejected at deployment creation with a clear validation error. Parameters are always validated again when a run is created. - Pydantic will coerce/validate types for the other fields (

int

,

bool

,

float

) and will apply any extra Field constraints you add (e.g.,

sampling_rate

0–1). Optional pattern: group parameters with a Pydantic model if you prefer a single validated config object.

from pydantic import BaseModel, Field

class FlowConfig(BaseModel):
    environment: str = "prod"
    mode: RunMode = RunMode.ITERATIVE
    window_hours: int = 1
    dry_run: bool = False
    sampling_rate: float = Field(1.0, ge=0.0, le=1.0)

@flow
def my_flow(config: FlowConfig):
    ...

YAML:

parameters:
  config:
    environment: prod
    mode: iterative
    window_hours: 1
    dry_run: false
    sampling_rate: 0.1

Docs for reference: - Flows and parameters: Flows (parameters) - Deployments and prefect.yaml: Deployments Quick check: Are you on Prefect 3.x? The above applies to 3.x. If you’re on 2.x, behavior differs and I can tailor the steps.

Looks like your message got cut off. Could you paste: - Your

RunMode

enum definition - The flow signature - The

deployments:

section from your

prefect.yaml

(with

parameters

and whether

enforce_parameter_schema

is set) - The exact error or behavior you’re seeing - Your Prefect version (3.x?) Quick checklist while you gather that: - Make your enum str-backed so YAML strings validate cleanly:

from enum import Enum
class RunMode(str, Enum):
    ITERATIVE = "iterative"
    BATCH = "batch"

- Annotate the flow parameter with the enum:

from prefect import flow

@flow  # validate_parameters=True by default
def my_flow(
    environment: str = "prod",
    mode: RunMode = RunMode.ITERATIVE,
    window_hours: int = 1,
    dry_run: bool = False,
    sampling_rate: float = 1.0,
):
    ...

- YAML parameters use plain scalars and straight quotes (not “smart quotes”):

deployments:
  - name: my-deployment
    entrypoint: path.to.module:my_flow
    parameters:
      environment: prod
      mode: iterative
      window_hours: 1
      dry_run: false
      sampling_rate: 0.1
    enforce_parameter_schema: true

If that still doesn’t validate, we can dump the inferred schema to confirm Prefect sees your enum:

from prefect.utilities.callables import parameter_schema
print(parameter_schema(my_flow.fn).model_dump())

Refs: - Flows (parameters) - Deployments

Thanks for the details — this looks close! A couple of key points to get enum validation (and avoid parameter errors): 1) Make sure the flow parameters match your YAML keys - Your YAML has a top-level

environment

, but your flow only accepts

config: FlowParams

. That means

environment

is an unexpected parameter and will fail validation when the run starts (or at deploy if schema enforcement is on). - Fix by either: - Adding

environment

to the flow signature, or - Moving

environment

into your

FlowParams

model. Option A — keep

environment

at top level:

from enum import Enum
from pydantic import BaseModel, Field
from prefect import flow

class RunMode(str, Enum):
    ITERATIVE = "iterative"
    BATCH = "batch"

class FlowParams(BaseModel):
    mode: RunMode = Field(default=RunMode.ITERATIVE, description="Processing mode")
    window_hours: int = Field(default=1, description="Size of the time window in hours")
    dry_run: bool = Field(default=False, description="Redirect output to test tables")
    sampling_rate: float = Field(default=1.0, ge=0.0, le=1.0, description="Fraction of data (0.01 = 1%)")

@flow(name="datapoint")
def datapoint_flow(
    environment: str = "dev",
    config: FlowParams = FlowParams(),
) -> int:
    ...

YAML:

parameters:
  environment: dev
  config:
    mode: iterative
    window_hours: 1
    dry_run: false
    sampling_rate: 0.1

Option B — move

environment

into the model:

class FlowParams(BaseModel):
    environment: str = "dev"
    mode: RunMode = Field(default=RunMode.ITERATIVE, description="Processing mode")
    window_hours: int = Field(default=1, description="Size of the time window in hours")
    dry_run: bool = Field(default=False, description="Redirect output to test tables")
    sampling_rate: float = Field(default=1.0, ge=0.0, le=1.0, description="Fraction of data (0.01 = 1%)")

@flow(name="datapoint")
def datapoint_flow(config: FlowParams = FlowParams()) -> int:
    ...

YAML:

parameters:
  config:
    environment: dev
    mode: iterative
    window_hours: 1
    dry_run: false
    sampling_rate: 0.1

2) Use a str-backed Enum for clean YAML validation and UI dropdowns

class RunMode(str, Enum):
    ITERATIVE = "iterative"
    BATCH = "batch"

- With this,

mode

will validate against ["iterative", "batch"], and the UI should render a dropdown. 3) Enforce schema in your deployment - Ensure your deployment has

enforce_parameter_schema: true

so invalid YAML values are rejected early. - After changing the flow signature/model, re-run your deployment so the schema updates and the UI dropdowns appear. If anything still isn’t validating, could you share: - Whether

enforce_parameter_schema

is set in your deployment - Any exact error you see (from deploy or run) - Your Prefect version (3.x?) Handy checks:

from prefect.utilities.callables import parameter_schema
print(parameter_schema(datapoint_flow.fn).model_dump())

Docs: - Flows and parameters: Flows (parameters) - Deployments (prefect.yaml): Deployments

Great — your

RunMode

enum is perfect for YAML-based validation:

class RunMode(str, Enum):
    ITERATIVE = "iterative"
    ON_DEMAND = "on_demand"
    BACKFILL = "backfill"

To ensure validation works end-to-end, two things to double-check: 1) Match your YAML parameters to your flow signature - Your YAML has a top-level

environment

, but your flow only accepts

config: FlowParams

. Unless you also accept

environment

in the flow signature, that top-level key will be rejected. - Either add

environment

to the flow signature, or move it into the

FlowParams

model. Option A — keep

environment

top-level:

from pydantic import BaseModel, Field
from prefect import flow

class FlowParams(BaseModel):
    mode: RunMode = Field(default=RunMode.ITERATIVE, description="Processing mode")
    window_hours: int = Field(default=1, description="Size of the time window in hours")
    dry_run: bool = Field(default=False, description="Redirect output to test tables")
    sampling_rate: float = Field(default=1.0, ge=0.0, le=1.0, description="Fraction of data (0.01 = 1%)")

@flow(name="datapoint")
def datapoint_flow(
    environment: str = "dev",
    config: FlowParams = FlowParams(),
) -> int:
    ...

YAML:

parameters:
  environment: dev
  config:
    mode: iterative
    window_hours: 1
    dry_run: false
    sampling_rate: 0.1

Option B — move

environment

into the model:

class FlowParams(BaseModel):
    environment: str = "dev"  # or an enum if you want
    mode: RunMode = Field(default=RunMode.ITERATIVE, description="Processing mode")
    window_hours: int = Field(default=1, description="Size of the time window in hours")
    dry_run: bool = Field(default=False, description="Redirect output to test tables")
    sampling_rate: float = Field(default=1.0, ge=0.0, le=1.0, description="Fraction of data (0.01 = 1%)")

@flow(name="datapoint")
def datapoint_flow(config: FlowParams = FlowParams()) -> int:
    ...

YAML:

parameters:
  config:
    environment: dev
    mode: iterative
    window_hours: 1
    dry_run: false
    sampling_rate: 0.1

2) Enforce schema at deploy/run time - In your deployment, set

enforce_parameter_schema: true

so invalid YAML values (e.g.,

mode: "wrong"

) are caught early. - Prefect will also validate again at run submission. An invalid enum will produce a clear error like:

mode: Input should be 'iterative', 'on_demand' or 'backfill'

. If you want to confirm the UI schema (and see the enum in it), you can print it:

from prefect.utilities.callables import parameter_schema
print(parameter_schema(datapoint_flow.fn).model_dump())

If something still isn’t validating, can you share: - Whether

environment

is in the flow signature or inside

FlowParams

- Whether

enforce_parameter_schema

is set - The exact error message - Your Prefect version (3.x) Refs: - Flows (parameters) - Deployments

Got it — you’re seeing the run form render

config

as a single JSON blob instead of expanded fields/dropdowns. That usually means the deployment’s stored parameter schema describes

config

as a generic object, not as your Pydantic model with defined properties/enums. Quick fixes and checks: 1) Re-deploy so the schema updates - Any change to the flow signature/model requires re-running your deployment so the

parameter_openapi_schema

is refreshed. - If you’re using prefect.yaml, run your deploy step again (e.g.,

prefect deploy ...

). 2) Verify the deployment’s parameter schema - Inspect the deployment to see how

config

is represented:

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

- In the printed

parameter_openapi_schema

, you should see something like:

"properties": {
  "config": { "$ref": "#/definitions/FlowParams" }
},
"definitions": {
  "FlowParams": {
    "type": "object",
    "properties": {
      "mode": { "type": "string", "enum": ["iterative", "on_demand", "backfill"] },
      "window_hours": { "type": "integer" },
      "dry_run": { "type": "boolean" },
      "sampling_rate": { "type": "number" }
    }
  }
}

- If

config

is just

"type": "object"

with no `$ref`/properties, the UI will fall back to a JSON editor. 3) Confirm Prefect and Pydantic versions - Prefect 3.x expects Pydantic v2. A mismatched/older Pydantic can cause incomplete schema generation.

python -c "import prefect, pydantic; print('prefect', prefect.__version__, 'pydantic', pydantic.__version__)"

4) Ensure the type annotation is resolvable at deploy time - Keep

config: FlowParams

as the annotation (no quotes/forward refs). - Make sure the

FlowParams

class is importable from the module referenced by your deployment

entrypoint

. 5) Optional vs required - The run form shows “(Optional)” if you provide a default value. If you want

config

required in the UI, don’t set a default:

@flow(name="datapoint")
def datapoint_flow(config: FlowParams) -> int:
    ...

- If you want it optional with defaults pre-filled in the UI, set

config: FlowParams = FlowParams()

. 6) Sanity check the schema directly from code

from prefect.utilities.callables import parameter_schema
print(parameter_schema(datapoint_flow.fn).model_dump())

- This should show

config

with

$ref

and a definition for

FlowParams

including the enum list. If after re-deploying you still get the JSON blob, please share: - Prefect version (3.x?) - Output of

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

- Output of

parameter_schema(datapoint_flow.fn).model_dump()

References: - Flows and parameters: Flows (parameters) - Deployments (prefect.yaml): Deployments

Thanks — that snippet is just the default parameter values. The UI decides how to render based on the deployment’s parameter_openapi_schema. If the schema says “config is a generic object”, the UI will show a single JSON blob. If the schema contains the FlowParams definition (with properties and the enum), the UI expands fields and renders a dropdown for mode. Let’s confirm what your deployment has stored: - Inspect the deployment and look specifically at parameter_openapi_schema

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

You should see something like:

'parameter_openapi_schema': {
  'properties': {
    'config': { '$ref': '#/definitions/FlowParams' }
  },
  'definitions': {
    'FlowParams': {
      'type': 'object',
      'properties': {
        'mode': { 'type': 'string', 'enum': ['iterative', 'on_demand', 'backfill'] },
        'window_hours': { 'type': 'integer' },
        'dry_run': { 'type': 'boolean' },
        'sampling_rate': { 'type': 'number' }
      }
    }
  }
}

If instead you see config just as

"type": "object"

(no $ref/properties), the UI will show JSON. Common causes and fixes: - The model wasn’t available at deploy time → Ensure FlowParams and RunMode are defined in the same module as the flow or are importable by the entrypoint, and then redeploy. - Schema not refreshed → Re-run your deploy so the stored schema updates after code changes. - Pydantic version mismatch → Prefect 3.x expects Pydantic v2. Check:

python -c "import prefect, pydantic; print('prefect', prefect.__version__, 'pydantic', pydantic.__version__)"

- Forward refs not resolved → Keep the annotation as config: FlowParams (not a quoted string). Define FlowParams and RunMode before the flow in the module. Local sanity checks (before redeploy):

from prefect.utilities.callables import parameter_schema
print(parameter_schema(datapoint_flow.fn).model_dump())

# Optional: confirm your model’s own schema includes the enum
print(FlowParams.model_json_schema())

If you paste the parameter_openapi_schema section from deployment inspect, I can pinpoint why the UI is falling back to JSON and suggest the exact fix.

Thanks — that schema explains the UI behavior. Your deployment’s parameter_openapi_schema only knows that “config” exists and has a default; it does not include the FlowParams type or its properties/enums. When the schema lacks type/details, the UI falls back to a single JSON blob. How to fix it: 1) Make sure Prefect can import your flow and model at deploy time - The CLI must import the entrypoint to generate the schema. If the code isn’t importable in the environment where you run prefect deploy (e.g., you’re using remote source code or missing PYTHONPATH), Prefect can’t introspect FlowParams and will store a minimal schema like the one you’re seeing. - Ensure: - Flow and FlowParams live in an importable module - The deployment entrypoint points to that module, e.g. path.to.module:datapoint_flow - Run prefect deploy from a venv where the package/module is installed or on the repo root with proper PYTHONPATH 2) Keep a resolvable annotation and default - Define FlowParams and RunMode before the flow in the same module (or cleanly importable). - Annotate and give a concrete default so it’s optional in the UI but still fully typed:

from enum import Enum
from pydantic import BaseModel, Field
from prefect import flow

class RunMode(str, Enum):
    ITERATIVE = "iterative"
    ON_DEMAND = "on_demand"
    BACKFILL = "backfill"

class FlowParams(BaseModel):
    mode: RunMode = RunMode.ITERATIVE
    window_hours: int = 1
    dry_run: bool = False
    sampling_rate: float = Field(1.0, ge=0.0, le=1.0)

@flow(name="datapoint")
def datapoint_flow(config: FlowParams = FlowParams()) -> int:
    ...

- Avoid forward-ref strings in the annotation (don’t write "FlowParams"). 3) Re-deploy to refresh the stored schema - After fixing the above, re-run your deploy so Prefect updates parameter_openapi_schema. 4) Verify the schema contains the model and enum - Check with:

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

- You should see something like:

'parameter_openapi_schema': {
  'type': 'object',
  'title': 'Parameters',
  'properties': {
    'config': { '$ref': '#/definitions/FlowParams' }
  },
  'definitions': {
    'FlowParams': {
      'type': 'object',
      'properties': {
        'mode': { 'type': 'string', 'enum': ['iterative', 'on_demand', 'backfill'] },
        'window_hours': { 'type': 'integer' },
        'dry_run': { 'type': 'boolean' },
        'sampling_rate': { 'type': 'number' }
      }
    }
  }
}

- If it’s still just:

'properties': {'config': {'title': 'config', 'default': None, 'position': 0}}

then Prefect still couldn’t import/resolve FlowParams at deploy time. 5) Quick local sanity check - Before deploying, confirm Prefect can build the schema in your dev env:

from prefect.utilities.callables import parameter_schema
print(parameter_schema(datapoint_flow.fn).model_dump())

You should see the $ref and FlowParams definition there too. 6) Version check (Prefect 3.x + Pydantic v2)

python -c "import prefect, pydantic; print('prefect', prefect.__version__, 'pydantic', pydantic.__version__)"

If you can share: - How you’re creating the deployment (CLI command or prefect.yaml deployments block, including entrypoint) - The output of parameter_schema(datapoint_flow.fn).model_dump() I can pinpoint what’s preventing schema generation in your deploy step.

Thanks — that schema confirms why the UI shows a single JSON blob. Your deployment stored a minimal schema: ‘parameter_openapi_schema’: { 'type': 'object', 'title': 'Parameters', 'properties': { 'config': { 'title': 'config', 'default': None, 'position': 0 } } } That means Prefect could not resolve `config`’s type at deploy time, so it treated it as

Any

. When the schema lacks the model/enum details, the UI falls back to a JSON editor. What to fix/check: 1) Ensure the deployed code actually annotates config with FlowParams and gives a concrete default - The stored schema shows

default: None

. That suggests your deployed flow signature is currently optional (e.g.,

config: FlowParams | None = None

) or FlowParams could not be resolved and Prefect defaulted to None. - Update the flow signature to this and re-deploy:

@flow(name="datapoint")
def datapoint_flow(config: FlowParams = FlowParams()) -> int:
    ...

2) Make sure Prefect can import the flow and model at deploy time - Prefect builds the schema by importing your entrypoint when you run

prefect deploy

. If the import fails or FlowParams cannot be resolved, the schema degrades to “any”. - Actions: - Define

FlowParams

and

RunMode

before the flow in the same module or import them normally (no conditional imports under TYPE_CHECKING). - Avoid forward-ref strings; use a real annotation

config: FlowParams

. - Ensure the code is importable in the environment where you run `prefect deploy`: - Add

__init__.py

files so

flows

,

flows/datapoint_age

, and

shared

are packages. - Use a module-style entrypoint in prefect.yaml (recommended):

flows.datapoint_age.<module>:datapoint_flow

instead of a file path. - Run

prefect deploy

from a venv where your project is importable (e.g.,

pip install -e .

or ensure PYTHONPATH points to your repo root). 3) Pin Prefect 3.x and Pydantic v2 - Prefect 3.x expects Pydantic v2. Verify:

python -c "import prefect, pydantic; print('prefect', prefect.__version__, 'pydantic', pydantic.__version__)"

- If you see pydantic 1.x, upgrade to

pydantic>=2,<3

and re-deploy. 4) Re-deploy to refresh the stored schema - After fixing code/imports, run your deploy again so

parameter_openapi_schema

updates. 5) Verify the deployment now has the model and enum in the schema

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

You should see:

'parameter_openapi_schema': {
  'type': 'object',
  'title': 'Parameters',
  'properties': {
    'config': { '$ref': '#/definitions/FlowParams' }
  },
  'definitions': {
    'FlowParams': {
      'type': 'object',
      'properties': {
        'mode': { 'type': 'string', 'enum': ['iterative','on_demand','backfill'] },
        'window_hours': { 'type': 'integer' },
        'dry_run': { 'type': 'boolean' },
        'sampling_rate': { 'type': 'number' }
      }
    }
  }
}

If you still see just “title/default/position”, Prefect is still not resolving your annotation at deploy time. Local sanity check (run in the same env you use for deploy):

from prefect.utilities.callables import parameter_schema
from flows.datapoint_age.<your_module> import datapoint_flow
print(parameter_schema(datapoint_flow.fn).model_dump())

- If this shows the $ref and definitions, the issue is with how you’re deploying/importing. - If it does not, the issue is in code ordering/imports/types. Your Dockerfile looks fine for runtime, but note: the UI form rendering depends on the schema stored at deployment creation, not on the Docker image. The critical part is that the environment running

prefect deploy

can import your flow and resolve

FlowParams

and

RunMode

. If you share: - The exact entrypoint you’re using in prefect.yaml - Your current flow signature line - Output of the local

parameter_schema(...).model_dump()

I can pinpoint the last mile. Helpful docs: - Flows and parameters: Flows (parameters)

- Deployments: Deployments