<@ULVA73B9P> what would be the best way to trigger...
# ask-marvin
f
@Marvin what would be the best way to trigger one flow based on the result of another flow?
m
thought for 252.6 seconds
Great question. Assuming you’re on Prefect 3.x, there are three solid patterns—pick based on how tightly you want the flows coupled and whether you need to pass data. - Simplest and most direct: call B as a subflow from A - Best when both flows can run in the same process and you want to branch on A’s Python return value. - You get A’s return value directly; no extra infra or waiting needed. - Orchestrator pattern: run deployments from code - Best when flows are deployed separately and may need different infrastructure. Use
run_deployment
and (optionally)
wait_for_flow_run
to conditionally kick off B based on A’s final state. - Event-driven: Automations - Best for loose coupling and “whenever A completes/fails, run B.” Create an Automation that listens for A’s flow-run events and performs a RunDeployment action to start B. - Note: Automations cannot read A’s Python return value directly. If you need to pass data, write it to an artifact, emit a custom event with a payload, or rely on parameters/tags. Code examples 1) Subflow pattern (tight coupling, access to return values)
Copy code
from prefect import flow

@flow
def flow_b(value: int) -> None:
    print(f"flow_b got {value}")

@flow
def flow_a() -> int:
    # ... do work and return a result
    return 42

@flow
def driver():
    a_result = flow_a()
    if a_result > 0:  # condition on A’s result
        flow_b(a_result)
2) Orchestrator using deployments (separate infra)
Copy code
from prefect import flow
from prefect.deployments import run_deployment
from prefect.flow_runs import wait_for_flow_run

@flow
def orchestrator():
    # Run A's deployment and wait for completion
    a_run = run_deployment(
        name="project/flow-a-deployment", 
        parameters={"param": "value"}, 
        as_subflow=False
    )
    a_run = wait_for_flow_run(a_run.id, log_states=True)

    # Trigger B only if A completed successfully
    if a_run.state_name == "Completed":
        b_run = run_deployment(
            name="project/flow-b-deployment",
            parameters={"upstream_id": str(a_run.id)},
            as_subflow=False
        )
        # Optionally wait for B as well
        wait_for_flow_run(b_run.id, log_states=True)
3) Event-driven with Automations (loose coupling) - Create an Automation that: - Trigger: Flow-run Completed (or Failed) for A (use tags or specific deployment to scope) - Action: RunDeployment → choose B’s deployment - Parameters: You can template with Jinja (e.g.,
{{ flow_run.id }}
,
{{ flow_run.parameters['foo'] }}
) If you need to pass A’s result to B via Automation: - Return values are not directly available to automations. Use one of: - Artifacts: A writes an artifact; B reads it. - Custom events: A emits a custom event with a payload; template in B’s parameters like
{{ event.payload.some_key }}
. - Parameters: If the “result” is known from A’s input parameters, pass them through to B with Jinja templates (e.g.,
{{ flow_run.parameters['source_date'] }}
). Helpful docs and references - Event-triggered automations overview: Event triggers - Creating deployment-triggering automations + templating: Creating deployment triggers and Access parameters in templates - Running deployments from code: Run deployments - Event-driven patterns and custom events: Event-driven flows with Prefect - Workflow patterns comparison: Workflow design patterns Quick recommendation - Use subflows if you just need “if A returns X, run B” and both can run together.
- Use
run_deployment
+
wait_for_flow_run
if A and B are separate deployments and you want code-controlled orchestration. - Use Automations if you want decoupled, event-driven behavior or to trigger B across multiple producers or schedules. A couple of clarifying questions to tailor this: - Do A and B need to run on different infrastructure or be decoupled? - Do you need to pass data from A to B, and if so, how large/structured is it? - Should B run on A Completed, Failed, or both?