Robert Szomolanyi
11/28/2025, 5:48 PMMarvin
11/28/2025, 5:48 PMMarvin
11/28/2025, 5:50 PMchild_flow()) will wait for it to finish.
- Use run_deployment(..., timeout=0) to return immediately after the child run is scheduled.
- Keep it linked as a “subflow” in the UI with as_subflow=True (default), or make it totally independent with as_subflow=False.
Examples
- Async parent flow (continues immediately, child stays linked as subflow):
from prefect import flow
from prefect.deployments import run_deployment
@flow
async def parent():
child_run = await run_deployment(
name="child_flow/child_deployment",
parameters={"x": 1},
timeout=0, # don't wait
as_subflow=True # keep parent/child link (default)
)
print(f"Started child run {child_run.id}")
# parent continues immediately
- Sync parent flow (also returns immediately; set as_subflow to False to make it independent):
from prefect import flow
from prefect.deployments import run_deployment
@flow
def parent():
child_run = run_deployment(
name="child_flow/child_deployment",
parameters={"x": 1},
timeout=0,
as_subflow=False # set False if you want no parent-child link
)
print(f"Started child run {child_run.id}")
# parent continues immediately
- If you prefer using the client directly:
from prefect import flow, get_client
@flow
async def parent():
async with get_client() as client:
fr = await client.create_flow_run_from_deployment(
deployment_id="YOUR-DEPLOYMENT-ID",
parameters={"x": 1},
# parent_task_run_id set automatically if you want it as a subflow;
# set None for independence
)
# parent continues immediately
Notes
- You must have a deployment for the child flow and a worker picking up its work.
- If by “subflow” you mean calling child_flow() directly, that will always wait—use a deployment trigger instead to avoid blocking.
Would you like me to show how to create the child deployment, or are you already set up with one?