XCom Data Management in the MCP Airflow API Server: Implementation and Usage

The MCP Airflow API server handles XCom data management through two asynchronous tools—list_xcom_entries and get_xcom_entry—that wrap the Airflow REST API with automatic authentication, pagination support, and JSON serialization.

The call518/mcp-airflow-api repository provides a Model Context Protocol (MCP) server that exposes Apache Airflow's operational data to Large Language Models. Central to this integration is robust XCom data management, enabling AI agents to inspect task outputs and inter-task communication without requiring direct database access.

Core XCom Data Management Tools

Located in src/mcp_airflow_api/tools/common_tools.py (lines 818-831), the server implements two primary methods for XCom interaction. Both are registered with the MCP framework using the @mcp.tool() decorator, making them discoverable to LLM clients.

Listing XCom Entries with Pagination

The list_xcom_entries function retrieves paginated XCom records for a specific DAG run and task. It constructs the endpoint:

GET /dags/{dag_id}/dagRuns/{dag_run_id}/taskInstances/{task_id}/xcomEntries

The function accepts limit (default 20) and offset (default 0) parameters to handle large XCom datasets efficiently. This pagination support is crucial for XCom data management when dealing with high-volume task outputs or historical DAG runs with numerous entries.

Retrieving Specific XCom Values

For targeted data retrieval, get_xcom_entry fetches a single XCom entry by its key:

GET /dags/{dag_id}/dagRuns/{dag_run_id}/taskInstances/{task_id}/xcomEntries/{xcom_key}

This method supports optional map indices for mapped tasks, providing granular access to specific XCom values without retrieving the entire dataset. The tool returns the raw JSON payload directly, including the key, value, and timestamp fields.

Authentication and Request Architecture

The XCom data management layer relies on the shared airflow_request utility in src/mcp_airflow_api/functions.py. This function handles:

1. URL Construction: Using construct_api_url to inject the configured API version and base URL
2. Authentication: Automatic selection between Basic Auth (Airflow v1) and JWT tokens (Airflow v2), with fallback to Basic Auth if token retrieval fails
3. HTTP Transport: Persistent aiohttp session management for asynchronous requests

When an XCom tool is invoked, it passes the constructed endpoint to airflow_request, which returns the JSON payload directly to the caller after raising for HTTP errors.

MCP Framework Integration

The XCom tools are fully integrated into the Model Context Protocol ecosystem through three key mechanisms:

Tool Registration: Both methods are decorated with @mcp.tool() in common_tools.py, exposing them to MCP clients.

Prompt Template Exposure: The src/mcp_airflow_api/prompt_template.md (lines 72-75) includes these tools under the XCom Management heading, ensuring Large Language Models understand when to invoke XCom data management operations during diagnostic workflows.

Diagnostic Workflows: In src/mcp_airflow_api/mcp_main.py (line 143), the main analysis routine explicitly references list_xcom_entries as the canonical method for examining XCom data when debugging DAG runs.

Practical Implementation Examples

Listing Paginated XCom Entries

from mcp_airflow_api.tools.common_tools import list_xcom_entries

# Retrieve first 20 XCom rows for task "process_data"

entries = await list_xcom_entries(
    dag_id="example_complex",
    dag_run_id="scheduled__2024-01-01T00:00:00+00:00",
    task_id="process_data",
    limit=20,
    offset=0
)

# Returns: {"xcom_entries": [...], "total_entries": 42}

Fetching a Specific XCom Value

from mcp_airflow_api.tools.common_tools import get_xcom_entry

# Retrieve specific XCom key "result"

entry = await get_xcom_entry(
    dag_id="example_complex",
    dag_run_id="scheduled__2024-01-01T00:00:00+00:00",
    task_id="process_data",
    xcom_key="result"
)

# Returns: {"key": "result", "value": "...", "timestamp": "..."}

LLM-Driven Diagnostic Workflow

When an LLM receives a request like "Show the XCom payload for the latest run of task data_processing in DAG example_complex", the MCP server executes:

1. Resolves the latest dag_run_id via the DAG runs tool
2. Calls list_xcom_entries with limit=1 to fetch the most recent entry
3. Returns the JSON payload to the LLM for user presentation

Summary

• The MCP Airflow API server implements XCom data management through two asynchronous tools: list_xcom_entries and get_xcom_entry in src/mcp_airflow_api/tools/common_tools.py.
• Both tools wrap the Airflow REST API with automatic authentication (Basic Auth or JWT) and pagination support (default limit 20) for handling large datasets.
• The tools are registered with the MCP framework via decorators and exposed in the prompt template, making them discoverable to Large Language Models.
• JSON serialization and persistent aiohttp sessions ensure efficient, asynchronous communication with the Airflow backend.

Frequently Asked Questions

What is XCom in Apache Airflow?

XCom (short for "cross-communication") is a mechanism in Apache Airflow that allows tasks to exchange small amounts of data. When a task pushes a value to XCom, subsequent tasks in the same DAG run can pull that value using the xcom_pull method. The MCP Airflow API server provides read-only access to this data through its XCom data management tools.

How does the server authenticate with the Airflow API?

The server uses the airflow_request function in src/mcp_airflow_api/functions.py to handle authentication automatically. For Airflow v1, it uses Basic Auth with the configured username and password. For Airflow v2, it attempts to obtain a JWT token first, falling back to Basic Auth if token retrieval fails. This authentication layer applies to all XCom data management operations.

Can I write or delete XCom entries using this MCP server?

No, the current implementation in call518/mcp-airflow-api provides read-only access to XCom data. The list_xcom_entries and get_xcom_entry tools only support HTTP GET requests to the Airflow REST API. To modify XCom entries, you would need to use the Airflow CLI, Python API, or direct database access outside of this MCP server.

What pagination limits should I use when listing XCom entries?

The list_xcom_entries tool defaults to limit=20 and offset=0, which is suitable for most debugging scenarios. However, if you expect large XCom payloads or numerous entries, you should implement pagination logic in your client code by incrementing the offset parameter in subsequent calls until all entries are retrieved. The response includes a total_entries field to help determine when pagination is complete.

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