Thanks for sharing the snippet. A few things could cause “inputs/parameters” issues at the start of a run, but I’ll need a bit more detail to pinpoint it. Could you share: - The exact error message/stack trace from the flow run or worker logs - Your flow function signature (the parameters your flow expects) - Prefect version (e.g., 3.x.y) - Whether you’re using

prefect.input.receive_input

or

RunInput

anywhere In the meantime, here are the most common causes and quick checks: - Parameter names must match your flow’s function signature exactly. - If your flow has required parameters and you schedule runs with

parameters={}

, you’ll get a ParameterBindError at start. Either pass values in

parameters=...

or give your flow defaults. - Parameters must be JSON-serializable (e.g., no raw datetimes or objects unless converted). - If you are using run inputs (

prefect.input.receive_input

), a scheduled run will pause/wait for input at the start of execution; confirm you intend to do that and that the code calls it within the flow/task (not at import time). - After changing your flow signature, make sure you re-deploy so the deployment’s parameter schema is updated. Here’s a minimal example of a working deployment with parameters and a cron schedule in Prefect 3.x:

from prefect import flow
from prefect.schedules import Cron

@flow
def client_flow(customer_id: int = 0, region: str = "EU"):
    # your logic here
    pass

if __name__ == "__main__":
    client_flow.from_source(
        source="<https://github.com/your-org/your-repo>",
        entrypoint="path/to/flows.py:client_flow",
    ).deploy(
        name="client",
        work_pool_name="workpool",
        tags=["PRO"],
        parameters={"customer_id": 123, "region": "EU"},  # must match flow signature
        schedule=Cron("0 20 * * *", timezone="Europe/Madrid"),
    )

If your flow has required parameters and you truly want to schedule with none, give them defaults in the flow signature. If you can paste the error and your flow signature, I can diagnose precisely and suggest the fix.

Thanks — I see your Pydantic model. I’m missing the exact error text, though. Could you paste the full error/stack trace from the run logs? That will let me confirm the root cause. Most likely issue - Deployment parameters are always JSON. If your flow expects an

ExperimentDefinition

but the deployment passes a JSON dict, your code will receive a dict and

exp.name_dir

will fail with "'dict' object has no attribute ..." unless you convert it to the model. Two ways to fix it: 1) Keep deployment parameters JSON and convert to the model inside the flow

from typing import Union
from pydantic import BaseModel, Field
from prefect import flow

class ExperimentDefinition(BaseModel):
    name_dir: str = Field(default="Segmentacion_de_Clientes")
    config_file: str = Field(default="config_pro.xlsx")
    col_name: str = Field(default="Produccion")

@flow
def client_flow(experiment: Union[ExperimentDefinition, dict] = None):
    # Coerce dict -> ExperimentDefinition
    exp = experiment if isinstance(experiment, ExperimentDefinition) else ExperimentDefinition(**(experiment or {}))
    # Use the model safely
    print(exp.name_dir, exp.config_file, exp.col_name)

Then in your deployment, pass a dict:

client_flow.from_source(...).deploy(
    name="client",
    work_pool_name="workpool",
    tags=["PRO"],
    parameters={
        "experiment": {
            "name_dir": "Segmentacion_de_Clientes",
            "config_file": "config_pro.xlsx",
            "col_name": "Produccion",
        }
    },
    schedule=Cron("0 20 * * *", timezone="Europe/Madrid"),
)

2) If you actually want interactive run inputs (human-in-the-loop) - Define your input as a RunInput and receive it at runtime. Note: scheduled runs will pause waiting for input.

from prefect import flow
from prefect.input import RunInput, receive_input
from pydantic import Field

class ExperimentDefinition(RunInput):
    name_dir: str = Field(default="Segmentacion_de_Clientes", title="Nombre del directorio")
    config_file: str = Field(default="config_pro.xlsx", title="Nombre del fichero config")
    col_name: str = Field(default="Produccion", title="Nombre de la columna")

@flow
def client_flow():
    exp = receive_input(ExperimentDefinition)  # pauses run until input is provided
    print(exp.name_dir, exp.config_file, exp.col_name)

A few extra checks - If you changed your flow signature recently, re-deploy so the parameter schema updates. - If you’re on Pydantic v2, use

.model_dump()

instead of

.dict()

when you need a dict. - If you keep parameters empty (

parameters={}

) but your flow expects required params, you’ll see a ParameterBindError. If you can share: - The exact error text - Your flow function signature - Prefect version (3.x.y) and Pydantic version I’ll tailor the fix precisely.

@Marvin I just want the code to use the inputs, experiment definition

Got it — if you just want your flow to accept an ExperimentDefinition “input” and use it, the simplest, robust pattern in Prefect 3.x is: - Keep deployment parameters JSON (dict) - Convert that dict into your Pydantic model at the start of the flow Here’s a drop-in example using your model.

from typing import Union
from pydantic import BaseModel, Field
from prefect import flow
from prefect.schedules import Cron

class ExperimentDefinition(BaseModel):
    name_dir: str = Field(default="Segmentacion_de_Clientes",
                          description="Nombre de la carpeta que se se va a verificar si existe.",
                          title="Nombre del directorio")
    config_file: str = Field(default="config_pro.xlsx",
                             description="Nombre del fichero de configuracion de los modelos de producción.",
                             title="Nombre del fichero config")
    col_name: str = Field(default="Produccion",
                          description="Nombre de la columna donde se aloja el nombre del modelo.",
                          title="Nombre de la columna")

@flow
def client_flow(experiment: Union[ExperimentDefinition, dict, None] = None):
    # Coerce JSON params -> Pydantic model with defaults
    exp = experiment if isinstance(experiment, ExperimentDefinition) else ExperimentDefinition(**(experiment or {}))

    # Use your inputs safely
    print(exp.name_dir, exp.config_file, exp.col_name)
    # ... your logic ...

# Deploy with JSON parameters; they’ll be converted in the flow
client_flow.from_source(
    source="<https://github.com/your-org/your-repo>",
    entrypoint="path/to/flows.py:client_flow",
).deploy(
    name="client",
    work_pool_name="workpool",
    tags=["PRO"],
    parameters={
        "experiment": {
            "name_dir": "Segmentacion_de_Clientes",
            "config_file": "config_pro.xlsx",
            "col_name": "Produccion",
        }
    },
    schedule=Cron("0 20 * * *", timezone="Europe/Madrid"),
)

Notes: - The deployment/UI “Parameters” field should contain a JSON object for

experiment

(keys must match your model fields). - If you leave

parameters={}

in the deployment, the flow will instantiate

ExperimentDefinition

with the defaults you defined. - After changing the flow signature, re-deploy so the parameter schema updates. If you prefer a human-in-the-loop form instead (runs wait for input), we can switch to RunInput, but for scheduled runs your job will pause. Otherwise, the pattern above is the right fit. If you share your current flow function signature, I can tailor this exactly to your code.

@Marvin i have parametes = {} and i do not know why the run is failing

If

parameters={}

causes the run to fail, the most common cause is that your flow has a required parameter. Even though your Pydantic model fields have defaults, the flow argument itself is still required unless you give it a default. Make the flow parameter optional and instantiate your model inside the flow so defaults apply when no params are passed. Option A (recommended: dict or None -> model)

from typing import Union
from pydantic import BaseModel, Field
from prefect import flow

class ExperimentDefinition(BaseModel):
    name_dir: str = Field(default="Segmentacion_de_Clientes",
                          description="Nombre de la carpeta que se se va a verificar si existe.",
                          title="Nombre del directorio")
    config_file: str = Field(default="config_pro.xlsx",
                             description="Nombre del fichero de configuracion de los modelos de producción.",
                             title="Nombre del fichero config")
    col_name: str = Field(default="Produccion",
                          description="Nombre de la columna donde se aloja el nombre del modelo.",
                          title="Nombre de la columna")

@flow
def client_flow(experiment: Union[ExperimentDefinition, dict, None] = None):
    # If nothing is passed (parameters={}), use model defaults
    exp = experiment if isinstance(experiment, ExperimentDefinition) else ExperimentDefinition(**(experiment or {}))
    # Use exp safely
    print(exp.name_dir, exp.config_file, exp.col_name)

Option B (keep typed param with a default instance)

@flow
def client_flow(experiment: ExperimentDefinition = ExperimentDefinition()):
    exp = experiment

Note: Option B uses a default object; it’s fine if you don’t mutate it. Option A is safer in general. Then your deployment can keep

parameters={}

and it will work:

client_flow.from_source(
    source="<https://github.com/your-org/your-repo>",
    entrypoint="path/to/flows.py:client_flow",
).deploy(
    name="client",
    work_pool_name="workpool",
    tags=["PRO"],
    parameters={},  # OK now
    schedule=Cron("0 20 * * *", timezone="Europe/Madrid"),
)

Important: - If you recently changed the flow signature, re-deploy so the deployment’s parameter schema updates. - If you’re using Prefect Run Inputs (

receive_input(ExperimentDefinition)

), a scheduled run will pause for input; that’s different from parameters and shouldn’t be mixed up. If this doesn’t resolve it, please paste the exact error/stack trace and your current flow function signature so I can pinpoint the issue.