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

hello prefect team,
i'm experiencing errors when trying to execute async flow deployments locally in prefect 2.0. how should i write my async flows so that i local execution works properly? running the flow script locally or running the deployment via an agent works perfectly fine but `prefect deployment execute ...`  fails with `RuntimeWarning: coroutine 'create_then_begin_flow_run' was never awaited` . see thread for my flow code. thanks a lot for your support.

```import asyncio
from prefect import flow, get_run_logger


@flow()
async def test_flow():
    logger = get_run_logger()
    <http://logger.info|logger.info>("Running test flow")


async def main():
    await test_flow()

if __name__ == "__main__":
    asyncio.run(main())```

I confirmed that it runs fine if not async and I get the same error if I run your code.

I’ll open an issue.

For your use case, can you just use a Deployment with an agent since that works fine?

Issue opened here: <https://github.com/PrefectHQ/prefect/issues/6060>