Data Lineage Tracking in Airflow API v2: Querying Asset Events with MCP

API v2 introduces data lineage tracking through the Assets model, replacing legacy datasets with asset events that record when DAGs create, update, or consume data assets.

The call518/mcp-airflow-api repository implements these capabilities as Model Context Protocol (MCP) tools, enabling programmatic access to Airflow 3.0+ lineage metadata. This article explains how to leverage the data lineage tracking capability in API v2 using the specific endpoints and parameters defined in the source code.

How Data Lineage Tracking Works in API v2

Airflow 3.0+ shifts from the legacy "datasets" paradigm to an Assets model for data-aware scheduling. Instead of querying dataset endpoints, v2 records asset events whenever a DAG creates, updates, or consumes an asset. This provides granular provenance tracking for data pipelines.

According to the source code in src/mcp_airflow_api/tools/common_tools.py, the legacy dataset-events helpers remain for backward compatibility but are superseded by asset-based tracking in v2. The new implementation resides in src/mcp_airflow_api/tools/v2_tools.py, where MCP tools wrap the Airflow REST API /assets/events endpoint.

Querying Asset Events with list_asset_events

The primary interface for lineage inspection is the list_asset_events tool, registered as an MCP tool in src/mcp_airflow_api/tools/v2_tools.py (lines 68‑71). This function queries the Airflow API for asset creation and update events, supporting pagination and filtering.

@mcp.tool()
async def list_asset_events(limit: int = 20, offset: int = 0,
                           asset_uri: Optional[str] = None,
                           source_dag_id: Optional[str] = None) -> Dict[str, Any]:
    """
    [V2 New] List asset events for data lineage tracking.

    Asset events track when assets are created or updated by DAGs.
    This enables data lineage tracking and data‑aware scheduling in Airflow 3.0.
    """

Key parameters:

• limit – Maximum events per page (default 20)
• offset – Pagination offset for large result sets
• asset_uri – Filter to events affecting a specific asset (e.g., s3://bucket/file.csv)
• source_dag_id – Filter to events produced by a specific DAG

Discovering Available Assets

Before querying events, use list_assets (also defined in v2_tools.py) to enumerate registered assets. This returns the data objects driving data-aware scheduling, allowing you to identify valid URIs for subsequent event filtering.

Practical Code Examples

The following examples demonstrate how to query lineage data using the MCP tool interface.

Example 1: List recent asset events with default pagination

await mcp.list_asset_events()

Example 2: Filter events for a specific data asset

await mcp.list_asset_events(asset_uri="s3://my-bucket/raw/orders.csv")

Example 3: Retrieve events generated by a specific DAG

await mcp.list_asset_events(source_dag_id="order_etl")

All calls return a standardized JSON-compatible dictionary:

{
  "asset_events": [...],
  "total_entries": 123,
  "limit": 20,
  "offset": 0,
  "api_version": "v2",
  "feature": "asset_events"
}

The asset_events array contains objects describing the operation type, timestamp, and participating DAGs, enabling full lineage reconstruction.

Implementation and Registration

When the MCP server initializes, it detects the configured API version and conditionally registers v2-specific tools. As implemented in src/mcp_airflow_api/mcp_main.py (lines 46‑49), the server exposes list_asset_events and list_assets only when operating against Airflow 3.0+ instances.

This dynamic registration ensures that clients cannot invoke asset-based lineage queries against legacy Airflow 2.x deployments that lack the underlying REST endpoints.

Summary

• API v2 replaces datasets with Assets for data lineage tracking in Airflow 3.0+, recording creation and consumption events rather than static dataset references.
• list_asset_events is the primary query endpoint, defined in src/mcp_airflow_api/tools/v2_tools.py, supporting filters by URI and source DAG with pagination.
• list_assets enables discovery of registered data objects before querying their event history.
• The MCP server dynamically registers these tools based on API version detection in mcp_main.py, ensuring compatibility with Airflow 3.0+ only.

Frequently Asked Questions

What replaced dataset events in Airflow API v2?

API v2 replaces the legacy dataset event system with asset events. While Airflow 2.x tracked datasets through the datasets REST endpoints, Airflow 3.0+ uses the Assets model where list_asset_events records when DAGs create, update, or consume data assets.

How do I filter asset events by source DAG?

Pass the source_dag_id parameter to list_asset_events. For example: await mcp.list_asset_events(source_dag_id="etl_pipeline"). This returns only asset events generated by tasks within that specific DAG, enabling targeted lineage analysis for individual pipelines.

What is the difference between list_assets and list_asset_events?

list_assets returns the catalog of registered data objects (the "what"), while list_asset_events returns the provenance log showing when DAGs interacted with those objects (the "when" and "who"). Use list_assets to discover available URIs, then query list_asset_events to track lineage and scheduling dependencies for specific assets.

Which Airflow version supports the asset-based lineage API?

The asset-based lineage tracking requires Airflow 3.0 or later. The MCP server detects the API version at startup and registers list_asset_events and list_assets only when connected to v2 endpoints, preventing compatibility errors with Airflow 2.x installations.

{
  "mcpServers": {
    "instagit": {
      "command": "npx",
      "args": ["-y", "instagit@latest"]
    }
  }
}