Great question. Short answer: don’t call

.result()

on not-yet-finished futures inside an async flow. Instead, iterate futures as they complete or use the async result methods so the event loop can keep heartbeats/lease TTL alive. Here are safe patterns in Prefect 3.x: Do this (stream results without blocking): - Use

prefect.futures.as_completed

to only call

.result()

on futures that are already finished. This makes

.result()

return immediately and avoids blocking the event loop.

from prefect import flow, task
from prefect.futures import as_completed

@task
def work(i):
    # some CPU or I/O work
    return i * 2

@flow
async def my_flow():
    futures = [work.submit(i) for i in range(10_000)]

    # SAFE: future.result() is only called when the future is done
    for fut in as_completed(futures, timeout=900):
        result = fut.result()  # instant, non-blocking here
        # you can do async work per result if needed
        # await some_async_side_effect(result)

Also safe (async-aware for remote runs): - If your futures are remote (workers) or flow-run futures, prefer their async result methods in an async flow: -

await fut.result_async(...)

for PrefectDistributedFuture (remote task runs) -

await fut.aresult(...)

for PrefectFlowRunFuture (flow runs)

@flow
async def my_async_flow():
    futs = [work.submit(i) for i in range(1000)]
    # Prefer async methods when available
    results = []
    for fut in as_completed(futs):
        if hasattr(fut, "result_async"):
            results.append(await fut.result_async())
        elif hasattr(fut, "aresult"):
            results.append(await fut.aresult())
        else:
            # Local/threaded futures don’t have async result;
            # but since as_completed yielded it, .result() is instant
            results.append(fut.result())
    return results

If you must wait for “all done” first: - Use

prefect.futures.wait(futures)

to get done/not_done sets, then call

.result()

on the done ones (instant, no block). - Or

PrefectFutureList(futures).result()

which resolves via an internal as-completed approach and preserves order. Anti-patterns that can block the event loop (risking missed heartbeat/lease TTL): - Calling

.result()

on futures that are not finished inside an async flow. - Polling states in a loop (e.g.,

while not fut.state.is_final(): await asyncio.sleep(0.1)

). - Sequential `.wait()`/`.result()` over many futures where some aren’t done yet. Optional pattern for mixed futures when you really want awaitables: - If you want to transform “not-yet-done” futures into awaitables uniformly, you can wrap non-async-result futures in

asyncio.to_thread

, but usually unnecessary if you use `as_completed`:

import asyncio

@flow
async def my_flow():
    futs = [work.submit(i) for i in range(1000)]
    awaitables = [
        fut.result_async() if hasattr(fut, "result_async")
        else fut.aresult() if hasattr(fut, "aresult")
        else asyncio.to_thread(fut.result)
        for fut in futs
    ]
    results = []
    for coro in asyncio.as_completed(awaitables):
        results.append(await coro)
    return results

Rule of thumb: - Inside async flows, never block the event loop waiting for work that isn’t done. - Prefer

as_completed

so

.result()

is instantaneous, or use the async result methods when available. If you can share whether your tasks run locally (ThreadPool) or on workers, and whether your flow is

async def

, I can tailor the example exactly to your setup.

You get a PrefectDistributedFuture in Prefect 3.x in two ways: 1) Use task.delay() (outside a flow) - Schedules the task for background execution and immediately returns a PrefectDistributedFuture. - Requires a running task worker to pick up and execute the task.

from prefect import task

@task
def add(x, y):
    return x + y

# Outside any flow
fut = add.delay(1, 2)  # <- PrefectDistributedFuture
res = fut.result()      # or: res = await fut.result_async()

2) Use PrefectTaskRunner in a flow and call .submit() - With PrefectTaskRunner, .submit() enqueues tasks like delay() and returns PrefectDistributedFuture, enabling awaitable result methods in async flows.

from prefect import flow, task
from prefect.task_runners import PrefectTaskRunner

@task
def work(i):
    return i * 2

@flow(task_runner=PrefectTaskRunner())
async def my_flow():
    futs = [work.submit(i) for i in range(100)]  # PrefectDistributedFuture[]
    # async-safe waiting
    results = [await f.result_async() for f in futs]
    return results

Don’t forget to run a task worker - Distributed tasks (from delay() or PrefectTaskRunner) won’t run unless a task worker is serving your tasks. CLI:

prefect task --help
prefect task serve --module my_package.tasks --limit 10
# or point to specific entrypoints
prefect task serve ./path/to/tasks.py:work

Python entrypoint (alternative):

# task_server.py
from prefect.task_worker import serve
from my_package.tasks import work, add

if __name__ == "__main__":
    serve(work, add, limit=10)

Notes - PrefectDistributedFuture supports async methods: await fut.result_async(...) and await fut.wait_async(...). - This is separate from flow workers/deployments; task workers specifically execute these distributed tasks.