Has anyone had trouble with required parameters and the

StartFlowRun

Task? I've been trying to run a flow of flows, but get validation errors despite passing the required parameters to the flow a mocked up example

param_a = Parameter(name="param_a")
param_b = Parameter(name="param_b", required=True)

@task(log_stdout=True)
def print_param(param):
   print(param)

# start flow run task instance
start_flow_run = StartFlowRun(project_name="required_test", wait=True)

with Flow("flow1") as f1:
  print_param(param_a)

with Flow("flow2") as f2:
  print_param(param_a)
  print_param(param_b) # the required one


with Flow("main") as f:
   # this works, ignoring the extra unused param
   start_flow_run(flow_name="flow1", parameters={"param_a": "a", "param_b": "b"}) 
   # this fails raising a validation error in the logs
   start_flow_run(flow_name="flow2", parameters={"param_a": "a", "param_b": "b"})

The error:

prefect.utilities.exceptions.ClientError: [{'path': ['create_flow_run'], 'message': "Required parameters were not supplied: {'param_b'}", 'extensions': {'code': 'INTERNAL_SERVER_ERROR'}}]

Hi @eli! I suspect the issue here is that the Parameters should be inside a Flow. Could you trying putting the

param_a

definition underneath flow1 and flow 2 and

param_b

underneath flow2?

Hey @eli -- that's a weird one indeed. That should be working. Could you run your script with ipython so you can hop in a debugger?

ipython -i your_script.py

then

%debug

after the error.

Your code is working for me

I figured out the issue. In my real world case because I was computing the parameters upstream in another task and waiting on a manual approval task, the results from the

compute_parameters

task were being "lost" (discarded) before start_flow_run was called.

Glad you figured it out! That’s good to know.