Using DAG Analysis Tools for Performance Optimization in the MCP Airflow API

The MCP Airflow API exposes Airflow's REST endpoints as callable Python coroutines through FastMCP tools like dag_run_duration, dag_task_duration, and dag_graph, enabling systematic identification of slow DAGs and inefficient tasks via automated workflows.

The call518/mcp-airflow-api repository bundles Airflow's administrative REST endpoints into a Model Context Protocol (MCP) server, making DAG analysis tools for performance optimization accessible as simple async function calls. These tools are registered at server start-up in src/mcp_airflow_api/mcp_main.py (lines 30-48) and grouped into common utilities compatible with both Airflow 2.x and 3.x, plus version-specific extensions. By leveraging these coroutines through Python clients, built-in prompts, or direct HTTP requests, you can automate the detection of runtime bottlenecks and structural inefficiencies across your entire DAG inventory.

Core DAG Analysis Tools

The MCP Airflow API organizes its instrumentation into common tools defined in src/mcp_airflow_api/tools/common_tools.py and version-specific tools for v2 endpoints in src/mcp_airflow_api/tools/v2_tools.py. For performance optimization, you'll primarily use the common toolkit, which includes list_dags, get_dags_detailed_batch, dag_run_duration, dag_task_duration, dag_graph, and dag_code.

All tools automatically utilize a pooled aiohttp session managed in src/mcp_airflow_api/functions.py (lines 18-46), ensuring connection reuse and timeout handling during large-scale analysis. The server exposes these through the FastMCP RPC layer, allowing any compatible client to invoke them directly.

The Performance Optimization Workflow

This systematic approach maps directly to tool invocations in common_tools.py:

1. Enumerate the DAG Inventory

Start with list_dags(limit, offset, fetch_all) (lines 55-73) to retrieve paginated DAG IDs and basic metadata. Setting fetch_all=True bypasses pagination for complete cluster coverage, ensuring no DAG escapes analysis.

2. Enrich with Detailed Metadata

Use get_dags_detailed_batch(limit, fetch_all, ...) (lines 89-111) to call list_dags_internal and augment each DAG with its latest run state via the /dagRuns?limit=1 endpoint. This provides the execution context needed to prioritize which DAGs warrant deep inspection.

3. Identify High-Duration DAGs

Call dag_run_duration(dag_id, limit) (lines 445-474) to compute per-run statistics including minimum, average, and maximum duration in seconds. Compare these averages across your inventory to reveal outliers consuming excessive scheduler or worker resources.

4. Drill Down to Task-Level Metrics

For slow DAGs, invoke dag_task_duration(dag_id, dag_run_id) (lines 485-511) to measure individual task execution times within a specific run. This isolates the precise bottleneck operator—whether it's a long-running SQL query, a heavy data transfer, or a misconfigured sensor.

5. Visualize Task Dependencies

Use dag_graph(dag_id) (lines 307-332) to retrieve upstream and downstream task relationships (via upstream_task_ids and downstream_task_ids). This exposes structural bottlenecks where long chains of sequential tasks delay completion, suggesting opportunities for parallelization.

6. Inspect Source Code

Retrieve the raw DAG definition with dag_code(dag_id) (lines 347-353) to check for misconfigured retries, heavy synchronous operators, or suboptimal schedule_interval settings that inflate runtime costs.

7. Monitor Active States

For real-time prioritization, running_dags() (lines 92-105) and failed_dags() (lines 27-42) provide snapshots of currently executing or problematic DAGs, allowing you to focus optimization efforts on active pain points.

Three Methods to Execute Analysis

Method 1: FastMCP Python Client

Connect via the stdio transport and invoke tools directly using the same coroutine names registered in the server:

import asyncio
from fastmcp import FastMCP

async def optimise():
    mcp = FastMCP("mcp-airflow-api")
    await mcp.connect()

    # Fetch full inventory

    dag_index = await mcp.list_dags(fetch_all=True)
    detailed = await mcp.get_dags_detailed_batch(fetch_all=True)

    # Identify top 5 slowest DAGs by average duration

    slow = []
    for dag in detailed["dags_detailed"]:
        stats = await mcp.dag_run_duration(dag["dag_id"], limit=20)
        avg = stats["statistics"]["average_duration_seconds"]
        slow.append((avg, dag["dag_id"]))
    
    slow.sort(reverse=True)
    top5 = [dag_id for _, dag_id in slow[:5]]

    # Analyze task-level bottlenecks

    for dag_id in top5:
        run_info = await mcp.dag_run_duration(dag_id, limit=1)
        recent_run_id = run_info["run_durations"][0]["run_id"]
        task_stats = await mcp.dag_task_duration(dag_id, dag_run_id=recent_run_id)
        
        bottlenecks = [
            t for t in task_stats["task_durations"]
            if t["duration_seconds"] > 300
        ]
        print(f"DAG {dag_id} bottlenecks:", bottlenecks)

    await mcp.close()

asyncio.run(optimise())

Method 2: Built-in DAG Analysis Prompt

The server registers three ready-made prompts in mcp_main.py (lines 54-84): airflow_cluster_monitoring, airflow_troubleshooting, and airflow_dag_analysis. Invoke the DAG analysis prompt via JSON-RPC to automate the workflow:

{
  "method": "airflow_dag_analysis",
  "params": {
    "analysis_type": "performance",
    "dag_pattern": "etl_*"
  }
}

This prompt automatically calls list_dags with id_contains="etl_", retrieves detailed batch info, runs dag_run_duration for each candidate, and summarizes the longest-running DAGs with suggested next steps.

Method 3: HTTP/CLI Ad-hoc Queries

Start the server in streamable-HTTP mode (python -m mcp_airflow_api --type streamable-http) to enable direct curl access:

# List currently running DAGs

curl -X POST http://localhost:8000/ \
  -d '{"method":"running_dags"}' | jq

# Get duration statistics for a specific DAG

curl -X POST http://localhost:8000/ \
  -d '{"method":"dag_run_duration","params":{"dag_id":"sales_report","limit":10}}' | jq

# Retrieve task-level timing for a specific run

curl -X POST http://localhost:8000/ \
  -d '{"method":"dag_task_duration","params":{"dag_id":"sales_report","dag_run_id":"manual__2024-01-01"}}' | jq

Performance Optimization Checklist

After gathering metrics, interpret symptoms using this diagnostic matrix:

Symptom	Likely Cause	Tool to Investigate
High average DAG run time	One or more tasks dominate execution	dag_task_duration → identify tasks > 5 minutes
Frequent retries in a DAG	Mis-configured retries or flaky external service	dag_code → inspect operator arguments
Long queue times before start	Over-subscribed pools or paused DAGs	list_pools / pause_dag / unpause_dag
DAGs never start after schedule	Scheduler lag or missing dependencies	dag_graph → visualize upstream bottlenecks
Sudden runtime spikes	Data-driven scheduling causing longer runs	list_asset_events (v2 only) → check recent asset churn

Key Source Files and Architecture

Understanding the codebase structure helps extend or debug the analysis tools:

File	Role	Key Components
src/mcp_airflow_api/tools/common_tools.py	Core DAG-analysis definitions	list_dags (L55-73), dag_run_duration (L445-474), dag_task_duration (L485-511), dag_graph (L307-332)
src/mcp_airflow_api/tools/v2_tools.py	v2-specific asset tools	list_assets, list_asset_events (L26-71) for Airflow 3.x data-driven scheduling
src/mcp_airflow_api/functions.py	Session management	Pooled aiohttp client with JWT auth for v2 (L18-46)
src/mcp_airflow_api/mcp_main.py	Server bootstrap	Tool registration (L30-48), prompt registration including airflow_dag_analysis (L54-84)
src/mcp_airflow_api/prompt_template.md	LLM workflow guidance	Human-readable prompt templates referenced by mcp_main.py

Summary

• The MCP Airflow API exposes Airflow's REST endpoints as FastMCP tools in common_tools.py, enabling programmatic DAG analysis for performance optimization.
• The seven-step workflow progresses from inventory (list_dags) to detailed metadata (get_dags_detailed_batch), duration analysis (dag_run_duration), task drill-down (dag_task_duration), dependency visualization (dag_graph), and code inspection (dag_code).
• Three access patterns support different use cases: Python FastMCP clients for custom automation, built-in prompts like airflow_dag_analysis for natural-language interaction, and HTTP endpoints for CLI ad-hoc queries.
• Connection pooling in functions.py ensures efficient batch analysis, while the v2 tools add asset-event tracking for modern data-driven scheduling issues.
• Systematic diagnosis using the symptom/cause matrix allows you to pinpoint whether slowness stems from task code, structural dependencies, resource pools, or scheduler configuration.

Frequently Asked Questions

How do I analyze DAG performance across multiple Airflow versions?

The common tools in src/mcp_airflow_api/tools/common_tools.py work with both Airflow 2.x (v1 API) and Airflow 3.x (v2 API). The server automatically detects the version at startup in mcp_main.py and registers compatible tools. For Airflow 3.x-specific features like data-driven scheduling, use the additional tools in v2_tools.py such as list_asset_events.

What is the difference between dag_run_duration and dag_task_duration?

dag_run_duration (lines 445-474) aggregates statistics across multiple DAG runs, returning min/avg/max durations to identify which DAGs are generally slow. dag_task_duration (lines 485-511) requires a specific dag_run_id and returns per-task execution times within that single run, allowing you to pinpoint exactly which operator in the pipeline consumes the most time.

Can I use these tools without writing Python code?

Yes. Start the server in streamable-HTTP mode and use curl or any HTTP client to POST JSON-RPC requests. Alternatively, invoke the airflow_dag_analysis prompt through any MCP-compatible client (such as Claude Desktop or other LLM interfaces) to run the optimization workflow using natural language commands.

How does the API handle authentication for large-scale analysis?

All tools use a shared aiohttp client session managed in src/mcp_airflow_api/functions.py (lines 18-46), which implements connection pooling, timeout handling, and automatic JWT token management for Airflow 3.x. This ensures that batch operations like get_dags_detailed_batch complete efficiently without exhausting connection limits or re-authenticating for every request.

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