airflow-hitl

Airflow Human-in-the-Loop Operators

Implement human approval gates, form inputs, and human-driven branching in Airflow DAGs using the HITL operators. These deferrable operators pause workflow execution until a human responds via the Airflow UI or REST API.

Implementation Checklist

Execute steps in order. Prefer deferrable HITL operators over custom sensors/polling loops.

CRITICAL: Requires Airflow 3.1+. NOT available in Airflow 2.x.

Deferrable: All HITL operators are deferrable—they release their worker slot while waiting for human input.

UI Location: View pending actions at Browse → Required Actions in Airflow UI. Respond via the task instance page's Required Actions tab or the REST API.

Cross-reference: For AI/LLM calls, see the airflow-ai skill.

---

Step 1: Choose operator

Operator	Human action	Outcome
ApprovalOperator	Approve or Reject	Reject causes downstream tasks to be skipped (approval task itself succeeds)
HITLOperator	Select option(s) + form	Returns selections
HITLBranchOperator	Select downstream task(s)	Runs selected, skips others
HITLEntryOperator	Submit form	Returns form data

---

Step 2: Implement operator

ApprovalOperator

from airflow.providers.standard.operators.hitl import ApprovalOperator
from airflow.sdk import dag, task, chain, Param
from pendulum import datetime

@dag(start_date=datetime(2025, 1, 1), schedule="@daily")
def approval_example():
    @task
    def prepare():
        return "Review quarterly report"

    approval = ApprovalOperator(
        task_id="approve_report",
        subject="Report Approval",
        body="{{ ti.xcom_pull(task_ids='prepare') }}",
        defaults="Approve",  # Optional: auto on timeout
        params={"comments": Param("", type="string")},
    )

    @task
    def after_approval(result):
        print(f"Decision: {result['chosen_options']}")

    chain(prepare(), approval)
    after_approval(approval.output)

approval_example()