Airflow Adapter

Airflow Adapter Pattern

Enables compatibility with both Airflow 2.x (/api/v1) and 3.x (/api/v2).

Architecture

MCP Tool → _get_adapter() → AirflowV2Adapter or AirflowV3Adapter → Airflow API

Version is auto-detected at startup.

Key Files

• adapters/base.py - Abstract interface
• adapters/airflow_v2.py - Airflow 2.x (/api/v1)
• adapters/airflow_v3.py - Airflow 3.x (/api/v2)

Related Files

• @api-differences.md - V2 vs V3 field/endpoint differences
• @patterns.md - Implementation patterns

Quick Reference

adapter = _get_adapter()
dags = adapter.list_dags(limit=100)
run = adapter.trigger_dag_run("my_dag", conf={"key": "value"})

Airflow API Differences (v2 vs v3)

Endpoint Paths

Field Name Changes

Authentication

Endpoints Only in Airflow 3.x

• dagStats without required dag_ids parameter
• Enhanced task instance filtering

Endpoints Changed in Airflow 3.x

• dagSources/{dag_id} - V3 uses dag_id directly, V2 needs file_token

Handling Differences in Code

# V2 adapter - normalize datasets to assets
def list_assets(self, ...) -> dict[str, Any]:
    data = self._call("datasets", ...)
    if "datasets" in data:
        data["assets"] = data.pop("datasets")
        for asset in data["assets"]:
            if "consuming_dags" in asset:
                asset["scheduled_dags"] = asset.pop("consuming_dags")
    return data

# V3 adapter - use native assets endpoint
def list_assets(self, ...) -> dict[str, Any]:
    return self._call("assets", ...)

Adapter Implementation Patterns

Basic Pattern

# In base.py
@abstractmethod
def get_thing(self, thing_id: str) -> dict[str, Any]:
    """Get a thing by ID."""
    pass

# In airflow_v2.py and airflow_v3.py
def get_thing(self, thing_id: str) -> dict[str, Any]:
    return self._call(f"things/{thing_id}")

Handling Missing Endpoints

def new_feature(self, param: str) -> dict[str, Any]:
    try:
        return self._call(f"newFeature/{param}")
    except NotFoundError:
        return self._handle_not_found(
            "newFeature",
            alternative="This feature requires Airflow 3.x"
        )

Field Normalization

When field names differ between versions, normalize in V2 to match V3:

# V2 adapter
def get_dag_run(self, dag_id: str, run_id: str) -> dict[str, Any]:
    data = self._call(f"dags/{dag_id}/dagRuns/{run_id}")
    # Normalize execution_date to logical_date
    if "execution_date" in data and "logical_date" not in data:
        data["logical_date"] = data["execution_date"]
    return data

POST/PATCH Operations

def trigger_dag_run(self, dag_id: str, conf: dict | None = None) -> dict[str, Any]:
    json_body: dict[str, Any] = {}
    if conf:
        json_body["conf"] = conf
    return self._post(f"dags/{dag_id}/dagRuns", json_data=json_body)

def pause_dag(self, dag_id: str) -> dict[str, Any]:
    return self._patch(f"dags/{dag_id}", json_data={"is_paused": True})

Password Filtering

Always filter passwords from connection data:

def list_connections(self, ...) -> dict[str, Any]:
    data = self._call("connections", ...)
    return self._filter_passwords(data)  # Base class method

Testing Adapters

def test_new_method(self, mocker):
    adapter = AirflowV2Adapter("http://localhost:8080", "2.9.0")

    mock_response = mocker.Mock()
    mock_response.json.return_value = {"result": "data"}
    mock_response.status_code = 200

    mock_client = mocker.Mock()
    mock_client.get.return_value = mock_response
    mock_client.__enter__ = mocker.Mock(return_value=mock_client)
    mock_client.__exit__ = mocker.Mock(return_value=False)

    mocker.patch("httpx.Client", return_value=mock_client)

    result = adapter.new_method("param")
    assert result == {"result": "data"}