Great question. In Prefect 3.x, the safest/recommended way to call async code from a synchronous task or flow is to use Prefect’s async interoperability utilities, primarily

run_coro_as_sync

. Recommended patterns 1) Sync task/flow calling async code → use

run_coro_as_sync

- Safely runs a coroutine from sync code without event-loop deadlocks - Prefect manages a dedicated event loop in a background thread Example:

from prefect import task, flow
from prefect.utilities.asyncutils import run_coro_as_sync
import httpx

@task
def fetch_user(user_id: int):
    async def _do():
        async with httpx.AsyncClient() as client:
            r = await client.get(f"<https://api.example.com/users/{user_id}>")
            return r.json()
    return run_coro_as_sync(_do())

@flow
def my_flow():
    return fetch_user(123)

2) One function usable in both sync and async contexts → use

@sync_compatible

- Lets you call the same function from sync code or

await

it from async code

from prefect.utilities.asyncutils import sync_compatible

@sync_compatible
async def compute(x: int) -> int:
    await some_async_work()
    return x * 2

# In a sync task:
# result = compute(5)

# In an async task:
# result = await compute(5)

3) Async task/flow calling blocking sync code → use

run_sync_in_worker_thread

- Prevents blocking the async event loop

from prefect.utilities.asyncutils import run_sync_in_worker_thread

@task
async def do_work():
    def blocking():
        time.sleep(2)
        return "ok"
    return await run_sync_in_worker_thread(blocking)

Key caveats and best practices - Do not use

asyncio.run

inside a task/flow. It can conflict with Prefect’s event loop management. Use

run_coro_as_sync

instead. - Avoid creating your own event loop or using libraries like

nest_asyncio

. - If most of the work is async I/O, prefer making the task itself

async def

and keep the call stack async. - Safe with Dask, process pools, and thread pools: each worker process/thread will manage its own loop state via Prefect’s utilities. Keep async operations self-contained within each task. - Reuse async clients (e.g., httpx) inside the coroutine you pass to

run_coro_as_sync

to benefit from connection pooling within the same worker process. References - Prefect async utilities source (includes

run_coro_as_sync

,

run_sync_in_worker_thread

,

sync_compatible

): prefect.utilities.asyncutils.py If you can share your specific scenario (e.g., calling an async SDK like httpx/aiohttp from a sync task, or your task runner/worker type), I can tailor an example and call out any environment-specific pitfalls.