Nikhil Jain
04/06/2026, 11:00 PMflow_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?Marvin
04/06/2026, 11:00 PMMarvin
04/06/2026, 11:03 PMasyncio.gather to run the two subflows concurrently, then continue once both finish.
Here’s a minimal pattern in Prefect 3.x:
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:
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.