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

> Optimize MCP Airflow API performance by using DAG analysis tools like dag_run_duration and dag_task_duration to systematically identify slow DAGs and inefficient tasks.

- Repository: [JungJungIn/mcp-airflow-api](https://github.com/call518/mcp-airflow-api)
- Tags: performance
- Published: 2026-02-26

---

**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`](https://github.com/call518/mcp-airflow-api/blob/main/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`](https://github.com/call518/mcp-airflow-api/blob/main/src/mcp_airflow_api/tools/common_tools.py) and **version-specific tools** for v2 endpoints in [`src/mcp_airflow_api/tools/v2_tools.py`](https://github.com/call518/mcp-airflow-api/blob/main/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`](https://github.com/call518/mcp-airflow-api/blob/main/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`](https://github.com/call518/mcp-airflow-api/blob/main/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:

```python
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`](https://github.com/call518/mcp-airflow-api/blob/main/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:

```json
{
  "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:

```bash

# 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`](https://github.com/call518/mcp-airflow-api/blob/main/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`](https://github.com/call518/mcp-airflow-api/blob/main/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`](https://github.com/call518/mcp-airflow-api/blob/main/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`](https://github.com/call518/mcp-airflow-api/blob/main/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`](https://github.com/call518/mcp-airflow-api/blob/main/src/mcp_airflow_api/prompt_template.md) | LLM workflow guidance | Human-readable prompt templates referenced by [`mcp_main.py`](https://github.com/call518/mcp-airflow-api/blob/main/mcp_main.py) |

## Summary

- **The MCP Airflow API** exposes Airflow's REST endpoints as FastMCP tools in [`common_tools.py`](https://github.com/call518/mcp-airflow-api/blob/main/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`](https://github.com/call518/mcp-airflow-api/blob/main/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`](https://github.com/call518/mcp-airflow-api/blob/main/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`](https://github.com/call518/mcp-airflow-api/blob/main/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`](https://github.com/call518/mcp-airflow-api/blob/main/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`](https://github.com/call518/mcp-airflow-api/blob/main/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.