Hi - has anyone had issues with a FlowRunTask not being added to a flow. I initially thought it was just because I was on prefect core, but even when registering with prefect cloud the task doesn’t show up in the schematics. I am also getting this warning each time…

UserWarning: Tasks were created but not added to the flow: {<Task: Flo Run Task Flow>}. This can occur when `Task` classes, including `Parameters`, are instantiated inside a `with flow:` block but not added to the flow either explicitly or as the input to another task.

Thank you

Hi @Shawn Marhanka - can you share your code?

Sure thing. I simplified it a bit.

from prefect import Flow, Parameter
from prefect.tasks.prefect.flow_run import FlowRunTask

...task imports

with Flow("Observe Data Flow") as observe_data_flow:
    uuid = Parameter("uuid")
    data = Parameter("data")
    observations_list = Parameter("observations")
    observer = Parameter("observer")
    callback_app = Parameter("callback_app")

    ...tasks translating data 
    merged_observation_list = flatten_observations([observations_list, other_task_result])

    next_flow_params = build_parameters(
        uuid,
        merged_observation_list,
        callback_app
    )
    trigger_flow_run = FlowRunTask(project_name="Project", flow_name="Flow Name",                                    parameters=next_flow_params)

Thanks for the quick reply

Great, thank you @Shawn Marhanka - can you try calling that task by adding

()

to the end like this:

trigger_flow_run = FlowRunTask(project_name="Project", flow_name="Flow Name", parameters=next_flow_params)()

Nice! That removed the warning and the task also shows up in the

observe_data_flow.tasks

set. However, when I run visualize(), it sets the FlowRunTask task off by itself. It isn’t downstream from the

next_flow_params

task like I would think it would be. The result of next_flow_params should feed the parameters of the FlowRunTask. Do you have to explicitly set that?

Nope, that should work as you expect, you'll just want to pass the parameters to the runtime call, in the

()

you just added, which will look something like this:

trigger_flow_run = FlowRunTask(project_name="Project", flow_name="Flow Name")(parameters=next_flow_params)

OH! gotcha. That worked. Thanks so much for the quick reply and solve!

Happy to help! For your knowledge, the reason that wasn't working as expected is because you were instantiating the

FlowRunTask

Class but never calling it, which is what Prefect looks for when building your Flow metadata

Makes sense. Thanks again. Have a great weekend.

You as well!