The Task Execution API in Airflow 3.0

Airflow 3.0 introduces a Task Execution API that decouples task execution from the scheduler core. A new Python Task SDK replaces the old direct coupling, and this architecture paves the way for SDKs in other languages (Go, Java).

The Python Task SDK

The Python Task SDK is a lightweight package (apache-airflow-task-sdk) that communicates with the scheduler via an HTTP API. Tasks run in isolated processes, improving stability and security.

python 
from airflow.sdk import DAG, task
from datetime import datetime


with DAG(
    dag_id='sdk_demo',
    schedule='@hourly',
    start_date=datetime(2025, 5, 1),
) as dag:

    @task
    def fetch_data() -> dict:
        """The SDK handles serialization automatically."""
        import requests
        resp = requests.get('https://api.example.com/data')
        return resp.json()

    @task
    def process(data: dict) -> list:
        """Tasks communicate via typed XCom."""
        return [item for item in data['results'] if item['active']]

    @task
    def store(records: list):
        print(f"Storing {len(records)} records")

    # Fluent chaining with the TaskFlow API
    store(process(fetch_data()))

# The SDK communicates with the scheduler via HTTP,
# no Airflow core imports needed in the worker.

Towards Multi-Language SDKs

Thanks to the standardized HTTP API, SDKs in other languages can define and execute Airflow tasks. A Go SDK is under development. This allows non-Python teams to integrate into Airflow pipelines.

go 
// Preview of the future Go SDK (under development)
package main

import (
    "fmt"
    airflow "github.com/apache/airflow-go-sdk"
)

func ProcessData(ctx airflow.TaskContext) error {
    data := ctx.GetXCom("fetch_data")
    fmt.Printf("Processing %d records\n", len(data))
    // Go processing for performance
    ctx.PushXCom("result", processed)
    return nil
}

// Go tasks communicate with the scheduler
// via the same HTTP API as the Python SDK.

Sources