MCP-Airflow-API Pagination: Handling 1000+ DAGs in Large-Scale Environments

MCP-Airflow-API implements server-side pagination through the list_dags_internal function in src/mcp_airflow_api/functions.py, supporting both manual offset-based navigation and automatic fetch_all aggregation for environments with thousands of DAGs.

Managing Apache Airflow deployments with 1000+ DAGs requires efficient pagination mechanisms to prevent timeouts and memory issues. The MCP-Airflow-API project addresses this challenge by implementing robust MCP-Airflow-API pagination logic that integrates seamlessly with both Airflow API v1 and v2. This article examines how the list_dags tool handles large-scale DAG enumeration through offset-based pagination and automatic fetching modes.

Core Pagination Logic in MCP-Airflow-API

The pagination implementation centers on the list_dags_internal function in src/mcp_airflow_api/functions.py. This function constructs Airflow API requests with limit and offset parameters, then processes the response to generate comprehensive pagination metadata.

Query Parameter Construction

When list_dags_internal receives a request, it builds the query string using the limit (page size) and offset (starting position) parameters. These values pass directly to the Airflow REST API's /dags endpoint, ensuring server-side filtering reduces network payload.

Pagination Metadata Calculation

After receiving the Airflow response, the function calculates three critical metadata fields:

• has_more_pages: Boolean indicating if additional pages exist, computed as (offset + limit) < total_entries
• next_offset: The offset value for the subsequent request (offset + limit when has_more_pages is true)
• pagination_info: Structured object containing current_page, total_pages, and remaining_count

This metadata enables clients to implement sequential pagination without tracking state externally.

Automatic Fetch-All Mode for Complete DAG Catalogs

For use cases requiring the complete DAG inventory, MCP-Airflow-API provides a fetch_all parameter that automates pagination traversal. When fetch_all=True, list_dags_internal enters a recursive loop:

1. Fetches the current page using the provided limit and current offset
2. Appends results to the accumulator
3. Checks has_more_pages to determine continuation
4. Updates offset to next_offset for subsequent iterations
5. Returns aggregated results when all pages are exhausted

The final response includes total_entries (filtered count), pages_fetched (iteration count), and the complete dags array. This mode is essential for LLM-driven workflows that require comprehensive DAG analysis without manual pagination handling.

Tool-Level Integration and API Version Compatibility

The pagination capabilities expose through the Model Context Protocol (MCP) tool interface, with architecture supporting both Airflow API versions.

Common Tools Implementation

The list_dags tool definition resides in src/mcp_airflow_api/tools/common_tools.py. This module imports the list_dags_internal function and registers it as an MCP tool with parameter schema matching the pagination options (limit, offset, fetch_all). The tool returns the structured response directly, maintaining consistency between the internal function and external API.

Version-Agnostic Architecture

MCP-Airflow-API supports both Airflow REST API v1 and v2 through a dependency injection pattern. The v1_tools.py and v2_tools.py modules set the appropriate airflow_request implementation before importing common_tools. This ensures the pagination logic in functions.py operates identically regardless of underlying Airflow version, as the HTTP transport layer abstracts version-specific endpoint differences.

Practical Implementation Examples

The following examples demonstrate pagination patterns for different operational scenarios.

Manual Pagination with Offset Tracking:

# Retrieve first page of 50 DAGs

result = await list_dags(limit=50, offset=0)
print(f"Retrieved {len(result['dags'])} DAGs")
print(f"More pages available: {result['has_more_pages']}")
print(f"Next offset: {result['next_offset']}")

# Request subsequent page using next_offset

if result['has_more_pages']:
    next_page = await list_dags(limit=50, offset=result['next_offset'])

Iterative Page Walking:

# Manually traverse all pages with custom processing per page

offset = 0
all_dags = []
page_count = 0

while True:
    page = await list_dags(limit=100, offset=offset)
    all_dags.extend(page["dags"])
    page_count += 1
    
    # Custom logic: process batch before continuing

    print(f"Processed page {page_count} ({len(page['dags'])} DAGs)")
    
    if not page["has_more_pages"]:
        break
    offset = page["next_offset"]

print(f"Total DAGs retrieved: {len(all_dags)} across {page_count} pages")

Automatic Fetch-All Mode:

# Retrieve complete DAG catalog in single call

# API handles pagination internally

full_catalog = await list_dags(fetch_all=True, limit=200)   # limit is per-page size

print(f"Total entries: {full_catalog['total_entries']}")
print(f"Pages fetched internally: {full_catalog['pages_fetched']}")
print(f"DAGs returned: {len(full_catalog['dags'])}")

# Process complete list without pagination logic

for dag in full_catalog["dags"]:
    print(f"DAG ID: {dag['dag_id']}, Status: {dag['status']}")

Summary

MCP-Airflow-API handles large-scale DAG environments through a robust pagination system implemented in src/mcp_airflow_api/functions.py. Key capabilities include:

• Offset-based pagination using limit and offset parameters with calculated metadata (has_more_pages, next_offset)
• Automatic aggregation via the fetch_all parameter that recursively retrieves all pages without client-side loop implementation
• Version compatibility through abstraction layers in v1_tools.py and v2_tools.py that inject appropriate HTTP clients before loading common tool definitions
• Structured metadata including total_entries, current_page, total_pages, and remaining_count to support UI pagination controls

Frequently Asked Questions

What is the default page size when listing DAGs in MCP-Airflow-API?

The default page size depends on the Airflow REST API configuration, but the list_dags tool accepts an explicit limit parameter to control page size. When using fetch_all=True, the limit parameter determines the batch size for each internal request, with common values ranging from 50 to 200 DAGs per page to balance memory usage and API call efficiency.

How does MCP-Airflow-API handle pagination when migrating between Airflow API v1 and v2?

The pagination logic remains identical across Airflow versions because src/mcp_airflow_api/functions.py operates on abstracted HTTP responses. The version-specific modules v1_tools.py and v2_tools.py set the appropriate airflow_request implementation before importing common_tools.py, ensuring that list_dags_internal receives consistent response structures regardless of whether the underlying Airflow deployment uses REST API v1 or v2.

Can I retrieve specific DAGs by ID while using pagination in MCP-Airflow-API?

Yes, the list_dags_internal function supports filtering parameters alongside pagination controls. When you specify filter criteria such as DAG ID patterns or tags along with limit and offset, the Airflow API applies filters before returning the paginated result set. For complete filtered catalogs, combine filter parameters with fetch_all=True to retrieve all matching DAGs across multiple pages while maintaining the filtering criteria throughout the aggregation process.

What happens if the Airflow API returns an error during paginated fetching?

When using manual pagination with individual list_dags calls, errors propagate immediately to the caller, allowing custom error handling for specific pages. When using fetch_all=True, the recursive loop in list_dags_internal raises the exception on the first failed request, halting the aggregation process. This ensures data integrity by preventing partial results from being returned as complete catalogs, though it requires callers to implement retry logic if fault tolerance is needed for large DAG environments.

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