<@ULVA73B9P> I want to call 2 async flows (say `fl...
# ask-marvin
n
@Marvin I want to call 2 async flows (say
flow_a
and
flow_b
) from inside another async flow (say
flow_c
). I want to kick off both
flow_a
and
flow_b
in parallel and once both are done, do some other work. How can I do this in prefect?
m
thought for 154.6 seconds
Yes—you can do this cleanly with an async parent flow and
asyncio.gather
to run the two subflows concurrently, then continue once both finish. Here’s a minimal pattern in Prefect 3.x:
Copy code
import asyncio
from prefect import flow

@flow
async def flow_a(x: int) -> int:
    await asyncio.sleep(1)  # stand-in for real work
    return x * 2

@flow
async def flow_b(y: int) -> int:
    await asyncio.sleep(2)  # stand-in for real work
    return y + 5

@flow
async def flow_c():
    # Kick off both subflows concurrently
    a_task = flow_a(10)
    b_task = flow_b(20)

    a_result, b_result = await asyncio.gather(a_task, b_task)

    # Both are done here — do other work
    combined = a_result + b_result
    return combined
Notes: - The parent must be async for this to run concurrently. -
asyncio.gather
raises if any subflow fails; if you want to collect both outcomes, use
return_exceptions=True
and handle them:
Copy code
results = await asyncio.gather(flow_a(10), flow_b(20), return_exceptions=True)
- Each subflow call creates its own flow run that you’ll see linked under the parent run in the UI. If instead you want to trigger registered deployments (so they can run on separate infrastructure and be cancelled/monitored independently), you can kick them off from the parent flow with
run_deployment
and manage them separately—but for simple in-process parallel subflows,
asyncio.gather
is the most straightforward. If that’s what you need, I can share a quick example for deployments too.