Hope this is the right place to ask this question. This might not be conventional and without going in too much detail about why but I basically created a special instance of the flow like this which the

_run

creates a new key by default inside the context object so it can be available to every task that is added to the flow

class MyFlow(prefect.Flow):
    def _run(
        self,
        parameters: Dict[str, Any],
        runner_cls: type,
        run_on_schedule: bool,
        **kwargs: Any
    ) -> "prefect.engine.state.State":
        context = dict(my_variable="my_variable")
        return super()._run(
            parameters,
            runner_cls,
            run_on_schedule=run_on_schedule,
            context=context,
            **kwargs
        )

This works as intended when I call

.run

and the task is able to access that key but it doesn’t work if I register the flow with a local ui and agent and try running it from the UI. Why is that and how would I make it work if I need to register a flow and run it from the UI?

@task(name="log-my-variable")
def log_run_id():
    prefect.context.get("logger").info(f"{prefect.context.my_variable}")


with MyFlow("experimental-flow") as flow:
    result = log_run_id()


if __name__ == "__main__":
    # flow.run()
    flow.register(project_name="test-flows")

Hey @Satyam Tandon, so

flow.run

is not used in the triggered runs by the UI. I think the agent uses a CloudFlowRunner so even if you have this class, the agent is still deploying the flow with the CloudFlowRunner so you might need to edit the agent here but it’s not a trivial process. I think our mechanism to deal with this is the KV Store on Prefect Cloud, but I have seen server users set up a Redis cache to pull this variable into their Flow.

Ah I see. Thanks, Kevin.

Yes all agents use the CloudFlowRunner

Gotcha. I have a few ideas I am going to try out but might have to bother you again. Thank your for your help and patience.

That

prefect.engine.cloud.CloudFlowRunner

is an import I believe so you would need to make it importable by the agent but not 100% sure

Hmm, I’m not sure what I need to do to make sure it’s importable by the agent? Currently the special class of FlowRunner I have is just in the working directory of the agent.

Like install it as a Python package maybe, but ideally not by editing Prefect cuz you kinda get locked in a version when you edit the Prefect source. I would package it as a Python package (with a setup.py) and pip install it.

I’m not sure where I would pip install it so it’s accessible to the agent? Right now that class is sitting where the flow code is in the working directory and I managed to change the environment variable

PREFECT__ENGINE__FLOW_RUNNER__DEFAULT_CLASS

on the local agent to the location of the python package that contains that special flow runner class

working_directory.<module_name>.<package_name>.<class_name>

but I don’t think it’s able to access it when the flow is registered and said agent is run pointed at localhost

You must have installed prefect on the agent with

pip install prefect

or something so it would be like that. You package the flow runner you made as a library and then do,

pip install the_new_library

.

Haha I thought of that but that wouldn’t work because the variable is actually effected by a user input (a prefect param) and I actually use the variable to overwrite the flow run name. So right now, that param has to be added for every flow we make and I’m trying to get away from that by doing everything in prefect context where I get the default value of that variable from just creating a flow object (so it would exist in prefect context to begin with and very importantly be available to every task created in the flow) and instead of having to add that param to every flow script, the user if they choose to not use the default simply specify the flow run name from the UI at run which you can without having to make any changes to any code as it that functionality you just get from the UI. That’s more context here. I don’t even know if this entire approach makes sense and if it will hit more roadblocks later but that’s what I’ve got so far thinking about this particular use case of mine.