You’re right to question that—`wait_for_flow_run` in Prefect 3.x is an async function. If you call it without awaiting, you’ll get exactly that error because you’re holding a coroutine object, not a

FlowRun

. What to know -

run_deployment(...)

returns a

FlowRun

synchronously. -

wait_for_flow_run(...)

is async and must be awaited. If you’re in synchronous code, wrap it with a helper like

run_coro_as_sync

or run it in an event loop. Relevant links - Source for `wait_for_flow_run`: prefect.flow_runs.wait_for_flow_run - How to run deployments programmatically: Run deployments (docs) - Async utils (includes

run_coro_as_sync

): prefect.utilities.asyncutils Examples 1) Pure synchronous code

from prefect.deployments import run_deployment
from prefect.flow_runs import wait_for_flow_run
from prefect.utilities.asyncutils import run_coro_as_sync

# Start the deployment (sync)
flow_run = run_deployment(
    name="my-flow/my-deployment",
    parameters={"x": 1},
    # optional knobs:
    # poll_interval=5,
    # timeout=600,
    # log_states=True,
)

# Wait for it to finish (bridge async -> sync)
flow_run = run_coro_as_sync(
    wait_for_flow_run(flow_run_id=flow_run.id, log_states=True)
)

if flow_run.state.is_completed():
    print("Success!")
else:
    print(f"Terminal state: {flow_run.state.type} - {flow_run.state.name}")

Alternative (sync) using asyncio directly:

import asyncio
from prefect.deployments import run_deployment
from prefect.flow_runs import wait_for_flow_run

flow_run = run_deployment(name="my-flow/my-deployment")
flow_run = asyncio.run(wait_for_flow_run(flow_run_id=flow_run.id))

Note:

asyncio.run

will fail if you’re already inside a running event loop (e.g., Jupyter). In those cases prefer

run_coro_as_sync

. 2) Inside async code

from prefect.deployments import run_deployment
from prefect.flow_runs import wait_for_flow_run

async def main():
    flow_run = run_deployment(name="my-flow/my-deployment", parameters={"x": 1})
    flow_run = await wait_for_flow_run(flow_run_id=flow_run.id, log_states=True)

    if flow_run.state.is_completed():
        ...

Why you saw

'coroutine' object has no attribute 'state'

-

wait_for_flow_run(...)

returns a coroutine until awaited. Accessing

.state

on that coroutine triggers the error. Always await it (or use

run_coro_as_sync

) before inspecting the resulting

FlowRun

. If you can share where you’re calling this from (plain script, within a flow, Jupyter, etc.), I can suggest the cleanest pattern for your setup.