Yes. In Prefect 3.x you can gate a subflow on upstream tasks. Two common patterns: 1) Call the subflow with wait_for - When you call a flow inside another flow (i.e., a subflow), you can pass

wait_for=[...]

with task futures. The subflow won’t start until those futures are complete.

from prefect import flow, task

@task
def extract():
    ...

@task
def transform():
    ...

@flow
def child_flow(payload: str):
    ...

@flow
def parent_flow():
    # kick off upstream tasks
    e = extract.submit()
    t = transform.submit()

    # subflow will not start until e and t are done
    subflow_future = child_flow(wait_for=[e, t], payload="ready!")
    result = subflow_future.result()

2) Use data dependencies (pass futures as inputs) - Passing task futures as parameters to the subflow also creates an upstream dependency; Prefect will wait for those futures before starting the subflow.

from prefect import flow, task

@task
def extract():
    return "data"

@task
def transform(x: str):
    return x.upper()

@flow
def child_flow(x: str, y: str):
    # runs only after both x and y are available
    ...

@flow
def parent_flow():
    e = extract.submit()
    t = transform.submit(e)

    # child_flow will wait on e and t before starting; it receives their results
    subflow_future = child_flow(x=e, y=t)
    subflow_future.result()

Triggering a deployment as a “subflow” - If you’re using

run_deployment

, it does not have a

wait_for

argument, but you can gate it the same ways: - Wait on futures before calling it - Wrap it in a task and use that task’s

wait_for

from prefect import flow, task
from prefect.futures import wait
from prefect.deployments import run_deployment

@task
def extract(): ...
@task
def preprocess(): ...

@task
def trigger_child_deployment(params: dict):
    return run_deployment("child-flow/production", parameters=params)

@flow
def parent_flow():
    e = extract.submit()
    p = preprocess.submit()

    # Option A: wrap in a task and use wait_for
    dep = trigger_child_deployment.submit({"k": "v"}, wait_for=[e, p])
    dep.result()

    # Option B: explicitly wait, then call inline
    wait([e, p])
    run_deployment("child-flow/production", parameters={"k": "v"})

Notes: - Tasks have a

wait_for

on

.submit(...)

. - Subflows (flow calls) have

wait_for

on the flow call itself. - Passing futures as parameters also creates dependencies automatically. If you can share whether you’re calling a code-defined subflow or triggering a deployment, I can tailor the example exactly to your setup.

Great question. When you use

run_deployment

, you’re launching a separate flow run (a deployment) and you don’t get a PrefectFuture directly. The best way to get the child flow’s return value is: - Create the flow run with

run_deployment(...)

- Wait for it to finish with

wait_for_flow_run(...)

- Extract the returned value from its terminal state with

get_state_result(...)

Here are two solid patterns: 1) Inline (blocks the parent until the child completes)

from prefect import flow
from prefect.deployments import run_deployment
from prefect.flow_runs import wait_for_flow_run
from prefect.states import get_state_result

@flow
def parent_flow():
    # Start the child deployment
    fr = run_deployment(
        "child-flow/production",
        parameters={"k": "v"},
        as_subflow=True,   # keeps lineage in the UI
    )
    # Wait for completion
    fr = wait_for_flow_run(fr.id, timeout=3600)
    # Get the child flow's return value (raises on failure by default)
    child_result = get_state_result(fr.state)
    return child_result

2) Wrap in a task to get a Future in the parent DAG - This lets you combine with

wait_for=[...]

or other task futures.

from prefect import flow, task
from prefect.deployments import run_deployment
from prefect.flow_runs import wait_for_flow_run
from prefect.states import get_state_result

@task
def trigger_and_get_child(params: dict):
    fr = run_deployment("child-flow/production", parameters=params, as_subflow=True)
    fr = wait_for_flow_run(fr.id, timeout=3600)
    return get_state_result(fr.state)

@flow
def parent_flow():
    # ... upstream work here, producing futures u1, u2, etc.
    fut = trigger_and_get_child.submit({"k": "v"})  # or pass wait_for=[u1, u2]
    child_result = fut.result()
    return child_result

Notes and gotchas: - Failures:

get_state_result(fr.state)

will raise if the child flow failed/crashed; catch exceptions if you want to handle them. - Serialization: The child flow’s return value must be serializable by your configured result serializer/storage. Simple Python types are fine. If you return large/complex objects, ensure result storage is configured and the serializer can handle it. - Lineage: Use

as_subflow=True

(default) so the run appears as a subflow of the parent in the UI.