Airflow TypeScript SDK

Public TypeScript interfaces for writing Apache Airflow task handlers.

Status: 0.1.0-beta1 · API may change · Node 22+ · ESM-only

This package defines the user-facing task handler contract and the coordinator runtime used to execute registered TypeScript handlers from Airflow.

Installation

npm install apache-airflow-ts-sdk@0.1.0-beta1

Task Handlers

import { Bundle, getClient, getContext, TaskHandler } from "apache-airflow-ts-sdk";

export async function sayHello() {
  const greeting = await getClient().getVariable("greeting");
  return { message: `Hello from ${getContext().taskId}: ${greeting}` };
}

const bundle = new Bundle();
bundle.register(new TaskHandler("example_dag", "say_hello", sayHello));
await bundle.serve();

A handler is a plain function. getContext() and getClient() reach the runtime from inside the call, so a handler takes no SDK-supplied parameter.

new TaskHandler(dagId, taskId, handler) binds the function to the Python-owned task it implements. The Dag is declared in Python with @task.stub, so the TypeScript side only supplies the task bodies.

Non-undefined return values are pushed to XCom under the "return_value" key by the active runtime, matching Python @task behavior.

Coordinator Usage

Airflow runs TypeScript task bundles through the Python-side airflow.sdk.coordinators.node.NodeCoordinator. Declaring Airflow Dags in TypeScript is not supported yet; the Dag is still declared in Python. The intended authoring shape matches the other non-Python SDKs: a Python Dag declares the scheduling shape with stub tasks, and the TypeScript module registers handlers with matching task IDs.

Python Dag:

from airflow.sdk import dag, task


@dag
def sales_pipeline():
    @task.stub(queue="typescript")
    def extract(): ...

    @task.stub(queue="typescript")
    def transform(extracted): ...

    transform(extract())


sales_pipeline()

Airflow coordinator config:

[dag_processor]
dag_bundle_config_list = [
    {"name": "dags-folder", "classpath": "airflow.dag_processing.bundles.local.LocalDagBundle", "kwargs": {}},
    {
      "name": "ts-task-handlers",
      "classpath": "airflow.dag_processing.bundles.local.LocalDagBundle",
      "kwargs": {"path": "/opt/airflow/ts-bundles"}
    }
  ]

[sdk]
coordinators = {
  "ts": {
    "classpath": "airflow.sdk.coordinators.node.NodeCoordinator",
    "kwargs": {"task_handler_bundle_name": "ts-task-handlers"}
  }
}
queue_to_coordinator = {"typescript": "ts"}

The Dag bundle named by task_handler_bundle_name (the task's own Dag bundle when unset) is searched recursively for *.min.mjs bundles built with airflow-ts-pack (see Packing bundles), which embeds the Airflow metadata in the bundle itself. task_handler_bundle_name is used only by mixed-language Dags, to locate the task handlers for the @task.stub tasks of a Python Dag; Dags defined natively in a language SDK do not use it.

This config, the packed bundles and Node.js 22 or later must be present wherever tasks execute and on the Dag processor, which checks the stub tasks of each Python Dag against the task handlers the bundles register. The API server does not need them.

TypeScript entrypoint:

import { Bundle, getClient, TaskHandler } from "apache-airflow-ts-sdk";

export async function extract() {
  const client = getClient();
  const connection = await client.getConnection("sales_db");
  const rowCount = Number((await client.getVariable("daily_row_count")) ?? "0");

  return {
    connectionId: connection?.id ?? null,
    rowCount,
  };
}

export async function transform() {
  const extracted = await getClient().getXCom<{ rowCount: number }>({
    key: "return_value",
    taskId: "extract",
  });

  return {
    transformedRows: extracted?.rowCount ?? 0,
  };
}

const bundle = new Bundle();
bundle.register(
  new TaskHandler("sales_pipeline", "extract", extract),
  new TaskHandler("sales_pipeline", "transform", transform),
);
await bundle.serve();

The Python stub defines the Dag dependency graph. The TypeScript handler does the work and uses TaskClient for task-time Airflow data access. The handler function is the reusable task implementation; a TaskHandler binds it to a Python stub task identity, a Bundle holds what this bundle process provides, and bundle.serve() serves it to Airflow.

bundle.serve() is the entrypoint: a task left unregistered is not part of the bundle, and one bundle can provide for several TaskHandlers. Registering holds no sockets and starts nothing, so a unit test can build a bundle and dispatch through bundle.getTaskHandler(dagId, taskId) without any runtime involved.

Dispatch keys on the (dagId, taskId) pair, so the same taskId under two Dags is two different handlers:

import { Bundle, TaskHandler } from "apache-airflow-ts-sdk";
import { chargeCustomer } from "./billing/tasks";
import { extract } from "./sales/tasks";

await new Bundle(
  new TaskHandler("sales_pipeline", "extract", extract),
  new TaskHandler("billing_pipeline", "extract", chargeCustomer),
).serve();

Register TaskHandler and Dag values with the register method, or pass them to the Bundle constructor.

TaskFlow arguments

A Python Dag that calls a stub task TaskFlow-style passes those arguments straight to the handler, which destructures them by name:

# the Python Dag
@task.stub(queue="typescript")
def transform(region_code: str, threshold: float, dry_run: bool = False): ...


transform("uk", 0.75)

interface TransformArgs {
  regionCode: string;
  threshold: number;
  dryRun: boolean;
}

export async function transform({ regionCode, threshold, dryRun }: TransformArgs) {
  // ...
}

Names bind by folding on both sides, lowercased with underscores removed, so region_code reaches regionCode and s3_uri reaches s3Uri with nothing declared on either side. An argument the call leaves at its default arrives with the default's value.

A name nothing folds to is logged, not thrown, naming what the handler asked for and what the call delivered. Two Python names that fold to the same token fail the task.

Object.keys and rest destructuring ({ ...rest }) yield Python's names, and in folds like a read.

Upstream outputs

An argument the Python call fills from another task arrives as that task's value, not as a reference to it. Feeding the transform above from an upstream task adds one argument on each side:

# the Python Dag
@task
def extract() -> int: ...


@task.stub(queue="typescript")
def transform(rows: int, region_code: str, threshold: float, dry_run: bool = False): ...


transform(extract(), "uk", 0.75)

interface TransformArgs {
  rows: number;
  regionCode: string;
  threshold: number;
  dryRun: boolean;
}

export async function transform({ rows, regionCode, threshold, dryRun }: TransformArgs) {
  // `rows` is the number extract() returned.
}

An upstream that pushed no output fails the task, naming both the argument and the task it came from. An upstream that pushed null binds null.

Being upstream is not the same as being passed. An XCom dependency declared with >> defines task order only, so a value the call did not pass is read explicitly:

const rows = await getClient().getXCom<number>({ key: "return_value", taskId: "extract" });

A Python int beyond the ±9007199254740991 a JavaScript number holds exactly is refused rather than bound, so carry such a value across the boundary as a string.

Explicit renames

withArgNames states a binding folding cannot reach, for a name the Python side never used: a clearer word than the Dag chose, or a TypeScript reserved word like enum. Mapping first, handler second:

interface ReportArgs {
  summary: Summary;
  label: string; // Python calls this `run_label`
}

const report = withArgNames({ label: "run_label" }, async ({ summary, label }: ReportArgs) => {
  // `label` is the call's `run_label`; `summary` folded as usual.
});

bundle.register(new TaskHandler("etl", "report", report));

An entry beats folding, and everything the map does not mention still folds, so withArgNames should be rare in a real Dag. The map‘s keys are checked against the handler’s own parameter type, so { labl: "run_label" } is a compile error naming the right key. Its values are Python names, which tsc cannot see and does not check.

Dag is another interface, for a Dag declared natively in TypeScript, and is still a work in progress.

Airflow launches the bundled entrypoint with --comm=host:port and --logs=host:port. bundle.serve() connects to those sockets, receives the task startup message, finds the registered handler for the Dag/task pair, and reports the terminal task state back to Airflow.

See example/ for a coordinator-runtime example that packs a bundle with airflow-ts-pack and uses a Python stub Dag.

Packing bundles

airflow-ts-pack produces a single self-contained bundle in one command. Packing is build-time only, so esbuild is an optional peer dependency the runtime install skips:

npm install --save-dev esbuild
airflow-ts-pack src/main.ts --outdir dist

It bundles the entrypoint into a minified dist/bundle.min.mjs with esbuild, then runs that bundle with --airflow-metadata so it reports its own registered Dag/task pairs and supervisor schema version. The manifest is embedded as a compact JSON //# airflowMetadata=... comment after a leading compact JSON //# airflowBundle=... layout descriptor, and the entry module is embedded verbatim in a /*# airflowSource ... #*/ block comment so Airflow can show the source a bundle was authored from, which its minified code no longer is. The CLI records the integrity metadata for all three regions in that descriptor, so a coordinator that is handed a bundle whose content was replaced fails loudly instead of running it. The result is one deployable file with no hand-written metadata sidecar.

Pass --outfile <path> instead of --outdir to name the artifact yourself, so one bundle directory can hold several bundles. The name must still end in .min.mjs, which is how NodeCoordinator finds bundles.

Options:

• --outdir <dir>: output directory (default dist)
• --outfile <path>: exact output path, whose name must end in .min.mjs
• --source <name>: display name of the primary source file shown in the Airflow UI (default: entry basename)

TaskClient

getClient() returns a TaskClient for task-time Airflow data access, for as long as a handler is running:

Locator fields such as dagId, runId, and taskId default to the current task context when omitted.

Cancellation

getContext().signal is an AbortSignal controlled by the active runtime. Pass it to fetch(), timers, database clients, child processes, or any other API that accepts an abort signal, so tasks can clean up cooperatively when Airflow terminates the task subprocess.

Compatibility matrix

Which Airflow TaskInstance states and capabilities this SDK supports. This table is generated from capabilities.yaml; the conformance dimensions are defined in the Language SDK conformance spec. Do not edit the table by hand. Update the manifest and run the update-ts-sdk-readme-matrix prek hook.

Min. Airflow version: 3.4 · supervisor schema: 2026-10-30

Marks: ✓ supported · ✗ not supported · n/a not applicable. A tier marked † applies only when native-dag-authoring is supported.

Links

• TypeScript SDK guide (staged docs) (how Airflow runs TypeScript task handlers)
• API reference (staged) (generated from the TypeScript sources)
• Source: the ts-sdk/ directory of the Apache Airflow monorepo
• Issues: bug reports and feature requests
• Website · Slack
• Developing this package (local build, docs, and the release workflow)