Passing Data Between Tasks: XComs and the TaskFlow API

Tasks run in separate processes, often on separate machines. XCom — cross-communication — is Airflow’s channel for passing small values between them.

The classic form

from datetime import datetime
from airflow import DAG
from airflow.operators.python import PythonOperator


def extract(**context):
    rows = [{"id": 1, "amount": 250}, {"id": 2, "amount": 180}, {"id": 3, "amount": 320}]
    print(f"extracted {len(rows)} rows")
    return {"row_count": len(rows), "path": "/tmp/extract_output.json"}


def transform(**context):
    # Pull the upstream return value by task_id.
    meta = context["ti"].xcom_pull(task_ids="extract")
    print(f"received: {meta}")

    total = 750
    context["ti"].xcom_push(key="revenue", value=total)
    return meta["row_count"] * 2


def load(**context):
    ti = context["ti"]
    doubled = ti.xcom_pull(task_ids="transform")                   # return_value
    revenue = ti.xcom_pull(task_ids="transform", key="revenue")    # explicit key
    print(f"doubled={doubled} revenue={revenue}")


with DAG(
    dag_id="xcom_classic",
    start_date=datetime(2026, 2, 1),
    schedule=None,
    catchup=False,
) as dag:
    t1 = PythonOperator(task_id="extract", python_callable=extract)
    t2 = PythonOperator(task_id="transform", python_callable=transform)
    t3 = PythonOperator(task_id="load", python_callable=load)
    t1 >> t2 >> t3

airflow tasks test xcom_classic extract   2026-02-14
airflow tasks test xcom_classic transform 2026-02-14
airflow tasks test xcom_classic load      2026-02-14

INFO - extracted 3 rows
INFO - Done. Returned value was: {'row_count': 3, 'path': '/tmp/extract_output.json'}

INFO - received: {'row_count': 3, 'path': '/tmp/extract_output.json'}
INFO - Done. Returned value was: 6

INFO - doubled=6 revenue=750

Two ways to push: returning a value (stored under the key return_value) or calling xcom_push with an explicit key. Two matching ways to pull.

The same DAG with TaskFlow

from datetime import datetime
from airflow.decorators import dag, task


@dag(
    dag_id="xcom_taskflow",
    start_date=datetime(2026, 2, 1),
    schedule=None,
    catchup=False,
)
def pipeline():

    @task
    def extract() -> dict:
        rows = [{"id": 1, "amount": 250}, {"id": 2, "amount": 180}, {"id": 3, "amount": 320}]
        print(f"extracted {len(rows)} rows")
        return {"row_count": len(rows), "path": "/tmp/extract_output.json"}

    @task
    def transform(meta: dict) -> int:
        print(f"received: {meta}")
        return meta["row_count"] * 2

    @task
    def load(doubled: int) -> None:
        print(f"doubled={doubled}")

    load(transform(extract()))


pipeline()

INFO - extracted 3 rows
INFO - received: {'row_count': 3, 'path': '/tmp/extract_output.json'}
INFO - doubled=6

Identical behaviour, and the dependency graph came from the function calls:

airflow tasks list xcom_taskflow --tree

<Task(_PythonDecoratedOperator): extract>
    <Task(_PythonDecoratedOperator): transform>
        <Task(_PythonDecoratedOperator): load>

No >> operators, no xcom_pull, no **context. Passing a task’s result as an argument both declares the dependency and moves the value. For new DAGs this is the default worth reaching for.

Mixing the two

TaskFlow tasks and classic operators compose:

from airflow.operators.bash import BashOperator

@dag(dag_id="mixed", start_date=datetime(2026, 2, 1), schedule=None, catchup=False)
def pipeline():

    @task
    def get_path() -> str:
        return "/data/2026-02-14/sales.csv"

    path = get_path()

    count_lines = BashOperator(
        task_id="count_lines",
        bash_command="wc -l < {{ ti.xcom_pull(task_ids='get_path') }}",
    )

    @task
    def report(p: str) -> None:
        print(f"processed {p}")

    path >> count_lines >> report(path)

pipeline()

INFO - Running command: ['/bin/bash', '-c', 'wc -l < /data/2026-02-14/sales.csv']
INFO - 1284
INFO - processed /data/2026-02-14/sales.csv

A BashOperator reads XComs through templating — {{ ti.xcom_pull(...) }} is rendered before the command runs.

Multiple outputs

Returning a dict and indexing it downstream creates one XCom for the whole dict. To split it:

    @task(multiple_outputs=True)
    def split() -> dict:
        return {"path": "/data/sales.csv", "rows": 1284}

    @task
    def use_path(p: str) -> None:
        print(f"path only: {p}")

    result = split()
    use_path(result["path"])

INFO - path only: /data/sales.csv

With multiple_outputs=True each key becomes its own XCom, so result["path"] pulls just that value instead of the whole dict. Without it, the subscript fails at parse time.

The size limit

XComs live in the metadata database. Try a large one:

    @task
    def too_big() -> list:
        return [{"id": i, "payload": "x" * 100} for i in range(100_000)]

[2026-02-14 11:31:07] {taskinstance.py:1938} ERROR - Task failed with exception
sqlalchemy.exc.DataError: (psycopg2.errors.ProgramLimitExceeded)
index row size 12384 exceeds btree version 4 maximum 2704 for index "xcom_pkey"

Even where it does not error outright, every XCom write and read is a database round trip. A DAG pushing megabytes through XCom will slow the scheduler for every other DAG on the cluster.

Pass a reference instead:

    @task
    def extract_to_storage() -> str:
        import json, tempfile, pathlib
        rows = [{"id": i, "amount": i * 10} for i in range(100_000)]
        path = pathlib.Path(tempfile.gettempdir()) / "extract.json"
        path.write_text(json.dumps(rows))
        print(f"wrote {len(rows)} rows to {path}")
        return str(path)                      # small string, not the data

    @task
    def process(path: str) -> int:
        import json, pathlib
        rows = json.loads(pathlib.Path(path).read_text())
        total = sum(r["amount"] for r in rows)
        print(f"read {len(rows)} rows, total {total}")
        return total

INFO - wrote 100000 rows to /tmp/extract.json
INFO - read 100000 rows, total 49999500000
INFO - Done. Returned value was: 49999500000

The rule: XComs carry pointers, storage carries data. In production that path is an S3 or GCS URI rather than local disk, since tasks may run on different machines.

Inspecting XComs

airflow xcoms list --dag-id xcom_classic 2>/dev/null || \
  airflow db shell <<'SQL'
SELECT dag_id, task_id, key, LEFT(CAST(value AS TEXT), 40) AS preview FROM xcom ORDER BY timestamp DESC LIMIT 5;
SQL

 dag_id       | task_id   | key          | preview
--------------+-----------+--------------+------------------------------------------
 xcom_classic | transform | revenue      | 750
 xcom_classic | transform | return_value | 6
 xcom_classic | extract   | return_value | {"row_count": 3, "path": "/tmp/extra

The UI shows the same under Admin → XComs. When a downstream task gets None, this is where to look — you will usually find the key is not what you expected.

Practice

    @task
    def silent() -> None:
        print("doing work")

    @task
    def reader(value) -> None:
        print(f"got: {value!r}")

    reader(silent())

INFO - doing work
INFO - got: None

context["ti"].xcom_push(key="rows", value=1284)
context["ti"].xcom_push(key="bytes", value=98304)
# downstream
ti.xcom_pull(task_ids="extract", key="rows")

1284

    data = extract()
    [transform_a(data), transform_b(data), transform_c(data)]

TypeError: Object of type DataFrame is not JSON serializable

Next: scheduling, backfill, and what the logical date really means.