Short answer: there’s no

merge

in Prefect 3.x. Instead, use the

wait_for

kwarg on your downstream task call and pass it the conditional task’s call result (or `None`/an empty list when the task didn’t run). Prefect will only create a dependency when there’s something to wait on. Pattern 1 — single optional dependency:

from prefect import flow, task

@task
def maybe_do_something():
    ...

@task
def downstream():
    ...

@flow
def my_flow(should_run: bool):
    maybe_dep = maybe_do_something() if should_run else None
    downstream(wait_for=maybe_dep)  # waits only if maybe_dep exists

Pattern 2 — accumulate zero-or-more dependencies:

from prefect import flow, task

@task
def step_a(): ...
@task
def step_b(): ...
@task
def downstream(): ...

@flow
def my_flow(run_a: bool, run_b: bool):
    deps = []
    if run_a:
        deps.append(step_a())
    if run_b:
        deps.append(step_b())
    downstream(wait_for=deps)  # works with [] or [one/many items]

Notes: - In Prefect 3.x,

Task.__call__

supports

wait_for

and accepts “one or many” futures/results. If you pass

None

or an empty list, there’s nothing to wait for; if you pass a task’s return (or future), it becomes a dependency. - This covers the old “merge” use case without needing a special node.