Hi, is it possible to use

prefect.engine.flow_runner.FlowRunner

or

prefect.engine.cloud.flow_runner.CloudFlowRunner

with a prefect server backend? I am wanting to run a flow and get task results so wanting to use the arg

return_tasks=[task1,task2]

. But I don't think I can do this with

prefect.Client().create_flow_run(version_group_id="redacted")

?

Hi @Ross - you should be able to use

CloudFlowRunner

for this;

CloudFlowRunner

existed before Prefect Server was a concept and so alludes to any Prefect API, rather than the Prefect Cloud product.

@nicholas ah ok thank you! I must have something set wrong in my config.toml file then because when I run:

fr = CloudFlowRunner(flow=flow)
flow_state = fr.run()
flow_state.result

I'm getting an error:

prefect.utilities.exceptions.ClientError('400 Client Error: Bad Request for url: <http://host>:port/graphql

(swapped out my host and port there) That why I had stupidly assumed this class was cloud only. I notice the url in the error is different to the one I use when successfully using

prefect.Client(<http://host>:port).register(flow)

in that it has

/graphql

added on the end... Not sure exactly where I am going wrong in the config file though. My simpified config just to the parts that I hope are relevant:

backend = "server"

[server]
host = "<http://host>"
port = "port"
host_port = "{server.port}"
endpoint = "${server.host}:${server.port}"

[cloud]
    api = "${${backend}.endpoint}"
    graphql = "${cloud.api}/graphql"

Any ideas?!

Hm, it looks like the

CloudFlowRunner

doesn't pass any arguments to the Client it creates, meaning the Client will take the endpoint directly from

config.cloud.graphql

- try changing that in your config to your server endpoint.

The flow runners aren't really public api - when running with a backend (e.g. Server/Cloud) you should let prefect setup and run the flow for you. If you want to view the results you'll want to set up

Result

objects (which will write the outputs out elsewhere) and view them later. The only supported public api for running a flow locally in a script (and then interacting with the results) is

flow.run()

.

Hmmm ok so I think

prefect.tasks.prefect.StartFlowRun

might be what I am after then for running/chaining flows. But that doesnt help me to get task results. Sounds like I need to do some more reading around the

Result

objects....! Thank you both