Great question. In Prefect 3.x, the dict you pass to

parameters

is just a mapping of parameter name to default value. Whether a parameter is required vs optional, and what its type is, comes from your flow function’s signature — not from the deployment API call itself. Here’s how it breaks down: - Passing a dict to parameters - You can pass a plain dict anywhere

parameters

is accepted (e.g.,

Flow.deploy(...)

,

run_deployment(...)

, or when building a

DeploymentCreate

for

client.create_deployment

). - Example using `Flow.deploy`:

from prefect import flow

    @flow
    def my_flow(required_id: int, threshold: float = 0.1, note: str | None = None):
        ...

    # Set deployment-level defaults for some params
    my_flow.deploy(
        name="prod",
        parameters={"threshold": 0.2, "note": "hello"},  # dict
        enforce_parameter_schema=True,  # validates types and requireds
    )

- Example when running a deployment:

from prefect.deployments import run_deployment

    run = run_deployment(
        name="my-flow/prod",
        parameters={"required_id": 42, "threshold": 0.15}
    )

- Optional vs required parameters - Required parameters are those without a default in the flow signature. - Optional parameters are those with a default value. If you set a default of

None

, also annotate as optional (e.g.,

str | None

or

Optional[str]

) to allow nulls. - Prefect infers a parameter schema from the flow’s signature and enforces it when

enforce_parameter_schema=True

(default in

Flow.deploy

). - Defining parameter types - Use standard Python type hints on your flow signature. Prefect will derive the schema from these annotations: - Primitives:

int

,

float

,

str

,

bool

- Containers:

list[str]

,

dict[str, int]

, etc. - Optional:

str | None

or

typing.Optional[str]

- Literals/enums for constrained values - Nested models: use a Pydantic model for rich validation of nested objects - Example with a Pydantic model:

from pydantic import BaseModel
    from typing import Optional
    from prefect import flow

    class Settings(BaseModel):
        retries: int = 3
        verbose: bool = False

    @flow
    def my_flow(
        required_id: int,                 # required (no default)
        threshold: float = 0.1,          # optional
        note: Optional[str] = None,      # optional, allows null
        settings: Settings = Settings(), # optional nested model
    ):
        ...

- Using PrefectClient.create_deployment directly -

create_deployment

takes a

DeploymentCreate

model. You include your

parameters

dict there. Optional vs required and typing are still determined by the flow’s signature; the deployment’s

parameters

just provide defaults. - Skeleton (omitting unrelated required fields for brevity):

from prefect.client.orchestration import get_client
    from prefect.client.schemas.actions import DeploymentCreate

    async def main():
        async with get_client() as client:
            dep = DeploymentCreate(
                # ... other required deployment fields (e.g., flow_id/entrypoint/work pool, etc.)
                parameters={"threshold": 0.2, "note": "hello"},
                enforce_parameter_schema=True,
            )
            await client.create_deployment(dep)

Tips and gotchas - Keep

enforce_parameter_schema=True

to have Prefect validate that defaults (and run-time overrides) match the types and requiredness derived from your flow. - Required parameters (no default) can be left out of

parameters

at deployment time; they must be supplied when you run the deployment. - If you pass

None

, make sure the type allows it (e.g.,

str | None

), otherwise validation will fail. Docs you may find useful: - How to create deployments: Create deployments

- CLI and Python deploy entrypoints (params and schema enforcement): Deploy how-to If you’d like a concrete

PrefectClient.create_deployment

example tailored to your flow, share your flow function signature and how you’re creating the deployment (flow source/entrypoint/work pool), and I’ll fill in the exact

DeploymentCreate

payload.

Short answer: Prefect inspects your flow’s Python signature and type hints, then builds a JSON/OpenAPI-style schema object. You can generate that same schema yourself and pass it to

client.create_deployment

via the

parameter_openapi_schema

field. How Prefect builds the schema - It uses

inspect.signature

to read parameter names, defaults (required vs optional), and type hints for each argument. - Types are converted into JSON Schema entries (e.g.,

int

->

{type: "integer"}

,

str

->

{type: "string"}

, optionals adjust

required

). - Docstrings are attached as field descriptions when available. - Nested/pydantic-typed params are referenced in

definitions

and `$ref`’d from

properties

. APIs you can call to generate it yourself - From a callable (recommended when you have the function in memory): -

prefect.utilities.callables.parameter_schema(fn)

returns a

ParameterSchema

object. - From an entrypoint string (recommended if you only know

"path/to/file.py:flow_func"

): -

prefect.utilities.callables.parameter_schema_from_entrypoint(entrypoint)

returns a

ParameterSchema

. - If you really want to build from primitives: -

prefect.utilities.callables.generate_parameter_schema(signature, docstrings)

. Each returns a

ParameterSchema

model which you can turn into a plain dict (for

parameter_openapi_schema

) with

.model_dump()

. Examples 1) Generate schema from a flow object and use in

create_deployment

from prefect import flow
from prefect.client.orchestration import get_client
from prefect.client.schemas.actions import DeploymentCreate
from prefect.utilities.callables import parameter_schema

@flow
def my_flow(required_id: int, threshold: float = 0.1, note: str | None = None):
    ...

async def main():
    # Build the OpenAPI-style schema dict from the flow's signature
    schema = parameter_schema(my_flow.fn).model_dump()

    async with get_client() as client:
        dep = DeploymentCreate(
            # ... include your other required deployment fields here ...
            # e.g., name, flow_id or entrypoint/source, work_pool, schedules, etc.
            parameter_openapi_schema=schema,
            enforce_parameter_schema=True,
            parameters={"threshold": 0.2}  # optional defaults at the deployment level
        )
        await client.create_deployment(dep)

2) Generate schema from an entrypoint string

from prefect.utilities.callables import parameter_schema_from_entrypoint

schema = parameter_schema_from_entrypoint("path/to/flows.py:my_flow").model_dump()
# Use `schema` in DeploymentCreate.parameter_openapi_schema as above

3) Optional: validate your schema and a proposed parameter set

from prefect.utilities.schema_tools import validate

# Raises with a helpful error if params do not conform to the schema
validate(schema, {"required_id": 123, "threshold": 0.3})

Notes and tips - Required vs optional comes from your flow signature: - No default => required (listed in

schema["required"]

) - Has a default => optional - If you want to allow

None

, annotate as

str | None

(or

Optional[str]

), not just

str = None

. - The dict you pass as

parameters

in a deployment just sets defaults. Validation of types/requiredness happens when

enforce_parameter_schema=True

. - If you have nested config, prefer Pydantic models in your flow signature; they’ll be converted into proper

$ref

definitions in the schema. Relevant docs - Create deployments: Create deployments - Validation tools overview: Schema tools and validation If you share your exact flow signature and how you’re constructing

DeploymentCreate

(flow source/entrypoint or flow_id), I can fill in a minimal, working payload tailored to your case.

Short answer: Prefect does not provide an AST-based generator. The built-in ways that Prefect generates the parameter schema load the flow (import/exec) and inspect its signature and type hints. If you truly cannot import the module, you can either: - Maintain a pre-built schema artifact (JSON) alongside your flow and pass it as

parameter_openapi_schema

to

client.create_deployment

, or - Roll your own best-effort AST extractor (brittle; won’t match Prefect’s full semantics for complex types). What Prefect provides today -

parameter_schema_from_entrypoint("path/to/file.py:flow_func")

will import/execute the entrypoint to produce a `ParameterSchema`:

from prefect.utilities.callables import parameter_schema_from_entrypoint

  schema = parameter_schema_from_entrypoint("flows.py:my_flow").model_dump()
  # use as DeploymentCreate.parameter_openapi_schema

- This mirrors what

prefect deploy

and

Flow.deploy

do under the hood. Recommended no-import pattern - Add a small CI/build step in the same repo as your flow that runs in a safe, isolated environment, imports the flow, and writes the schema to a JSON file. Then your “deployment creator” can read that JSON and pass it to

parameter_openapi_schema

without importing the flow at runtime. - Because this uses Prefect’s own generator, it stays accurate with complex annotations, Optional/Union, Pydantic models, and docstrings. If you still want to use AST (best-effort example) Caveats: - You will miss runtime-evaluated/forward-referenced annotations, imports, Pydantic model expansion, Literals, custom encoders, etc. - Defaults that are expressions won’t be easily serializable. - You’ll need to duplicate a subset of Prefect’s type mapping. Here is a minimal AST-based extractor that handles common primitives and simple generics to a JSON Schema-like dict Prefect accepts as `parameter_openapi_schema`: ``` import ast from pathlib import Path PRIMITIVES = { "int": {"type": "integer"}, "float": {"type": "number"}, "str": {"type": "string"}, "bool": {"type": "boolean"}, "Any": {}, "None": {"type": "null"}, } def _name(node): if isinstance(node, ast.Name): return node.id if isinstance(node, ast.Attribute): return f"{_name(node.value)}.{node.attr}" if hasattr(ast, "unparse"): return ast.unparse(node) return None def _annotation_to_schema(node): # Primitive names n = _name(node) if n in PRIMITIVES: return PRIMITIVES[n] # Optional[X] / Union[X, None] / X | None if isinstance(node, ast.Subscript): base = _name(node.value) # typing.Optional[...] or Optional[...] if base and base.split(".")[-1] in ("Optional", "Union"): # Extract args args = [] if isinstance(node.slice, ast.Tuple): elts = node.slice.elts else: elts = [node.slice] for elt in elts: args.append(_annotation_to_schema(elt)) # If any is null, union with null types = [] for a in args: t = a.get("type") if isinstance(t, list): types.extend(t) elif t: types.append(t) else: types.append(None) if "null" in types: types = [t for t in types if t] # drop None markers types = list(dict.fromkeys(types)) # unique if types: return {"type": types + ["null"]} else: return {"type": "null"} # list[T] / List[T] if base and base.split(".")[-1] in ("list", "List"): item_schema = _annotation_to_schema(node.slice) return {"type": "array", "items": item_schema} # dict[K, V] / Dict[K, V] if base and base.split(".")[-1] in ("dict", "Dict"):

if isinstance(node.slice, ast.Tuple) and len(node.slice.elts) == 2: _, v = node.slice.elts return {"type": "object", "additionalProperties": _annotation_to_schema(v)} else: return {"type": "object"} # PEP 604 unions: X | Y if isinstance(node, ast.BinOp) and isinstance(node.op, ast.BitOr): left = _annotation_to_schema(node.left) right = _annotation_to_schema(node.right) lt = left.get("type") rt = right.get("type") union_types = [] for t in (lt, rt): if isinstance(t, list): union_types.extend(t) elif t: union_types.append(t) union_types = list(dict.fromkeys(union_types)) or None if union_types: return {"type": union_types} return {} # Fallback unknown annotation return {} def generate_parameter_schema_from_file(path: str, func_name: str): src = Path(path).read_text(encoding="utf-8") mod = ast.parse(src, filename=path) func = next((n for n in mod.body if isinstance(n, ast.FunctionDef) and n.name == func_name), None) if not func: raise ValueError(f"Function {func_name} not found in {path}") # Build properties and required based on defaults args = func.args.args # positional-only not handled here; extend as needed defaults = func.args.defaults or [] num_required = len(args) - len(defaults) required = [a.arg for a in args[:num_required]] properties = {} for i, a in enumerate(args): ann = a.annotation schema = _annotation_to_schema(ann) if ann is not None else {} properties[a.arg] = schema return { "title": "Parameters", "type": "object", "properties": properties, "required": required, "definitions": {}, } # Example usage: schema = generate_parameter_schema_from_file("flows.py", "my_flow") # Pass

schema

as DeploymentCreate.parameter_openapi_schema

Again, this is intentionally limited:
- It won’t expand Pydantic models into `$ref` + `definitions`.
- It won’t resolve imported type aliases, TypedDicts, Literals, constrained types, or docstrings.
- It won’t match Prefect’s behavior for advanced annotations.

Using the schema with client.create_deployment
- Once you have the dict, you can pass it as `parameter_openapi_schema` and set `enforce_parameter_schema=True`. Example skeleton:

from prefect.client.orchestration import get_client from prefect.client.schemas.actions import DeploymentCreate async def create_dep(schema: dict): async with get_client() as client: dep = DeploymentCreate( # ... your other required fields (name, flow/source/entrypoint/work pool, etc.) parameter_openapi_schema=schema, enforce_parameter_schema=True, parameters={}, # optional defaults at deployment level ) await client.create_deployment(dep) ``` If you share your constraints (why import is not possible) and your target flow annotations (e.g., do you use Pydantic models?), I can suggest a more reliable approach or extend the AST snippet to cover your specific cases. For most production setups, I recommend the CI-generated schema artifact approach to avoid surprises while still keeping your deployment path import-free.