How to trigger a DAG from another DAG with parameters in Airflow

TriggerDagRunOperator starts a run of another DAG and sends it a dictionary of values, the conf. The triggered DAG declares the values it accepts as params, with defaults and types, so the same DAG can also run on its own or from the UI form.

Copy and paste

The DAG that triggers:

from airflow.providers.standard.operators.trigger_dagrun import TriggerDagRunOperator

trigger_export = TriggerDagRunOperator(
    task_id="trigger_export",
    trigger_dag_id="export_sales",
    conf={"region": "us", "full_refresh": True, "run_date": "{{ ds }}"},
)

The DAG that is triggered:

import pendulum
from airflow.sdk import dag, task, Param

@dag(
    schedule=None,
    start_date=pendulum.datetime(2026, 1, 1, tz="UTC"),
    params={
        "region": Param("eu", type="string", enum=["eu", "us", "apac"]),
        "full_refresh": Param(False, type="boolean"),
        "run_date": Param(None, type=["null", "string"]),
    },
)
def export_sales():
    @task
    def export(params=None):
        print(params["region"], params["full_refresh"], params["run_date"])

    export()

export_sales()

In a templated field the same values are {{ params.region }}, {{ params.run_date }} and so on.

Wait until the triggered run has finished:

trigger_export = TriggerDagRunOperator(
    task_id="trigger_export",
    trigger_dag_id="export_sales",
    conf={"region": "us"},
    wait_for_completion=True,
    poke_interval=30,
    deferrable=True,   # frees the worker slot while waiting; needs the triggerer
)

How the values arrive

conf is a templated field, so "{{ ds }}" becomes the date of the triggering run. The triggered DAG receives the dictionary as dag_run.conf, and every key it sends replaces the default of the matching param. Keys you do not send keep their default: with the conf above, full_refresh would be False if it were left out.

Params also validate the input. A value outside the enum, such as "region": "mars", fails the validation and the run does not start.

The logical date of the triggered run

Unless you pass logical_date, the triggered run gets the moment of the trigger as its logical date, not the date of the DAG that triggered it. Its ds is therefore today. That is why the example sends the business date as run_date in the conf.

You can pass logical_date="{{ logical_date }}" instead, but a DAG accepts one run per logical date. Triggering the same date a second time, for example when the calling task is retried, fails with a message that the run already exists. Two parameters handle that case:

• reset_dag_run=True clears the existing run and runs it again;
• skip_when_already_exists=True marks the trigger task as skipped.

Other details

• With wait_for_completion=True the task succeeds only if the triggered run succeeds. Change that with allowed_states and failed_states.
• If the triggered DAG is paused, its run waits in the queue until someone unpauses it.
• When the second DAG should run because the first one produced new data, rather than because the first one says so, assets are an alternative: the first DAG declares an asset as output and the second is scheduled on it.

Tested with Airflow 3.3.2 and the standard provider 1.19.