Bring your towel and join one of the fastest growing data communities. Welcome to our second-generation open source orchestration platform, a completely rethought approach to dataflow automation.

Prefect Community

Hi, I am trying to run an async flow and get: AttributeError: 'coroutine' object has no attribute 'is_failed'. Any thoughts?

Could you elaborate a bit more? What would be helpful is:
1. Share the output of `prefect version`
2. Share flow we could use to reproduce the error
3. Share the traceback you're getting

Any async flow, even the simplest one triggers the error, Prefect 2.0b6,  `File "/home/user/.local/share/virtualenvs/deploytest-gAiUW3yO/lib/python3.9/site-packages/prefect/cli/deployment.py", line 153, in execute`
    `if state.is_failed():`
`AttributeError: 'coroutine' object has no attribute 'is_failed'`

I couldn't replicate - async flow from the docs
```import asyncio

from prefect import task, flow


@task
async def print_values(values):
    for value in values:
        await asyncio.sleep(1)  # yield
        print(value, end=" ")


@flow
async def async_flow():
    await print_values([1, 2])  # runs immediately
    coros = [print_values("abcd"), print_values("6789")]

    # asynchronously gather the tasks
    await asyncio.gather(*coros)


if __name__ == "__main__":
    asyncio.run(async_flow())```
logs
```13:36:32.779 | INFO    | prefect.engine - Created flow run 'gray-bird' for flow 'async-flow'
13:36:32.780 | INFO    | Flow run 'gray-bird' - Using task runner 'ConcurrentTaskRunner'
13:36:32.786 | WARNING | Flow run 'gray-bird' - No default storage is configured on the server. Results from this flow run will be stored in a temporary directory in its runtime environment.
13:36:32.812 | INFO    | Flow run 'gray-bird' - Created task run 'print_values-94c2122e-0' for task 'print_values'
13:36:32.829 | INFO    | Flow run 'gray-bird' - Created task run 'print_values-94c2122e-1' for task 'print_values'
13:36:32.849 | INFO    | Flow run 'gray-bird' - Created task run 'print_values-94c2122e-2' for task 'print_values'
1 a 6 2 b 7 13:36:34.846 | INFO    | Task run 'print_values-94c2122e-0' - Finished in state Completed()
c 8 d 9 13:36:36.850 | INFO    | Task run 'print_values-94c2122e-1' - Finished in state Completed()
13:36:36.866 | INFO    | Task run 'print_values-94c2122e-2' - Finished in state Completed()
13:36:36.877 | INFO    | Flow run 'gray-bird' - Finished in state Completed('All states completed.')```


it runs fine if you try it outside of a deployment, if you create a deployment and try to execute it then it fails

image.png

I did and it looked fine

```DeploymentSpec(name="local", flow=async_flow)```

could you open a GitHub issue with a detailed explanation of the flow you tried to run, how did you trigger it and share a full traceback? might be easier to reproduce and potentially fix then since I can't reproduce based on the current information set you provided

if I try prefect deployment execute it fails, if I try prefect deployment run it indeed dispatches the run

same as in your screenshot, however, it is marked as late in the gui and never runs

Ok my bad, I had not started an agent. please mention this explicitly in the documentation as it is not particularly obvious. I had thought that starting prefect automatically starts an agent! Thank you <@U02H1A95XDW>

awesome, glad you figured that out

regarding your feedback, let me cc <@U02FDNUGQ3H> if we can improve that somehow in the docs