Hi team, I am currently struggling registering flo...
# marvin-ai
α
Hi team, I am currently struggling registering flows from
controlflow
inside Prefect Cloud. Is this functionality supported ? Am I overseeing something ? Thank you in advance.
j
I dont expect there to be any issues running flows in Prefect Cloud - can you provide any more detail?
α
Hi jeremiah, thanks for the reply. I will show you 2 cases, one that works (meaning that after running the script, I can see the flow in Prefect Cloud ) and one that doesn't.
By running the following code 'my_flow.py'
Copy code
from prefect import flow, task

@task
def task_one():
    print("Hello from task 1")

@task
def task_two():
    print("Hello from task 2")

@task
def task_three():
    print("Hello from task 3")

@flow
def flow_that_registers():
    task_one()
    task_two()
    task_three()

if __name__ == "__main__":
    flow_that_registers()
by just hitting
python my_flow.py
, I can see the run on Prefect Cloud. See the screenshot.
But then, running this:
Copy code
import controlflow as cf

@cf.flow
def flow_that_doesnt(x):
    y = cf.run('Add 5 to the number', result_type=int, context=dict(x=x))
    z = cf.run('Multiply the result by 2', result_type=int)
    return z

if __name__ == "__main__":
    flow_that_doesnt(100)
with
python my_controlflow.py
, the script executes, but I see nothing in Prefect Cloud.
Is that expected behaviour ?
j
That is not expected!
I will do some testing - perhaps we inadvertently overwrote something
thank you for highlighting
1
α
Thank you for taking it into consideration. Please keep me posted on any updates on this, as we'd love to move our production agentic workflows in Prefect ASAP. Meanwhile I will continue testing and let you know if anything of substance comes up.
I tried to deploy the flow, instead of just running it.
Copy code
from pathlib import Path
import controlflow as cf

@cf.flow
def flow_that_doesnt(x):
    y = cf.run('Add 5 to the number', result_type=int, context=dict(x=x))
    z = cf.run('Multiply the result by 2', result_type=int)
    return z

if __name__ == "__main__":
    flow_that_doesnt.from_source(
        source=str((p := Path(__file__)).parent.resolve()),
        entrypoint=f"{p.stem}.py:flow_that_doesnt",
    ).deploy(name="my-first-cf-deployment",
                      tags=["cf-pls-run"],
                      parameters={"elemeiooooooo": True},
                      work_pool_name="wp1",
                      interval=10*60)
I am getting an error that the work pool cannot be detected, even though I have the exact workpool with the exact name. Part of the error is :
Copy code
prefect.exceptions.PrefectHTTPStatusError: Client error '404 Not Found' for url '<http://127.0.0.1:8471/api/work_pools/wp1>'
Response: {'detail': 'Work pool "wp1" not found.'}
Does this mean that controlflow is not looking at my cloud, but tries to detect locally the workpools ? i have authenticated with
prefect cloud login
, is there any additional
controlflow cloud login
that I should perform ?
j
Ok - I think this is fixed here https://github.com/PrefectHQ/ControlFlow/pull/311. This PR was open for a while but facepalm never got merged, for exactly this reason. At the time P3 was new and it was the opposite - Cloud was running but not local, so we had added overrides to help it discover the local server. I think now that's overeager and causing the issue -- I believe this should work