MCP Tools for Apache Airflow — DAG runs, task instance states, and REST API

Authentication and API client setup

Airflow's REST API uses Basic Authentication by default on self-hosted deployments. The credentials are the same as the Airflow web UI admin account. Managed Airflow services (Google Cloud Composer, Amazon MWAA, Astronomer) add service-account-based authentication: Cloud Composer uses Google IAP tokens, MWAA uses SigV4-signed requests, and Astronomer uses API tokens. For local and development deployments, Basic Auth with admin credentials is standard.

The API base URL is typically https://<airflow-host>/api/v1. Airflow 2.x serves the REST API from the webserver component, not the scheduler — the webserver must be reachable for API calls even if the scheduler is running. Check GET /api/v1/health (no auth required) to confirm webserver availability before making authenticated calls.

class AirflowClient {
  constructor({ baseUrl, username, password }) {
    this.baseUrl = baseUrl.replace(/\/$/, '');
    this.authHeader = `Basic ${Buffer.from(`${username}:${password}`).toString('base64')}`;
  }

  async request(method, path, body = null) {
    const url = `${this.baseUrl}/api/v1${path}`;
    const headers = {
      'Authorization': this.authHeader,
      'Content-Type': 'application/json',
      'Accept': 'application/json',
    };

    const response = await fetch(url, {
      method,
      headers,
      body: body ? JSON.stringify(body) : undefined,
    });

    if (response.status === 401) throw new Error('Airflow auth failed — check credentials');
    if (response.status === 403) throw new Error('Insufficient Airflow permissions');
    if (response.status === 404) throw new Error(`Not found: ${path}`);
    if (!response.ok) {
      const text = await response.text();
      throw new Error(`Airflow API error ${response.status}: ${text}`);
    }

    if (response.status === 204) return null; // No content (DELETE, some PATCHes)
    return response.json();
  }

  // Check webserver health (no auth required)
  async checkHealth() {
    const url = `${this.baseUrl}/api/v1/health`;
    const response = await fetch(url);
    return response.json();
    // Returns: { metadatabase: { status: 'healthy' | 'unhealthy' },
    //            scheduler: { status: 'healthy' | 'unhealthy', latest_scheduler_heartbeat: '...' } }
  }
}

DagRun state machine and triggering runs

A DagRun has five possible states: queued (created but scheduler hasn't picked it up), running (scheduler is executing tasks), success (all tasks completed successfully), failed (at least one task failed and no retry will rescue the run), and dataset_triggered (a special state for dataset-driven DAGs). Only success and failed are terminal states; queued and running require polling.

The logical_date parameter is Airflow's primary identifier for a DAG run within a schedule. It represents the start of the data interval this run processes — for a daily DAG scheduled at midnight, the logical_date 2026-07-11T00:00:00Z processes data from July 11. If you trigger a run with a specific logical_date and a run for that slot already exists, Airflow returns a 409 Conflict. To force re-execution, delete the existing run first or use a unique run_id with the same logical date. For on-demand triggering without a schedule slot, you can omit logical_date and Airflow sets it to the current UTC time.

async function triggerDagRun(client, dagId, options = {}) {
  const body = {};

  if (options.logicalDate) {
    body.logical_date = options.logicalDate; // ISO8601 string
  }

  if (options.conf) {
    body.conf = options.conf; // JSON dict accessible via {{ dag_run.conf }} in tasks
  }

  if (options.runId) {
    body.dag_run_id = options.runId; // Custom run ID (must be unique per DAG)
  }

  const dagRun = await client.request('POST', `/dags/${dagId}/dagRuns`, body);

  return {
    dagRunId: dagRun.dag_run_id,
    dagId: dagRun.dag_id,
    logicalDate: dagRun.logical_date,
    state: dagRun.state, // 'queued' immediately after trigger
    startDate: dagRun.start_date,
    executionDate: dagRun.execution_date, // Alias for logical_date (Airflow 2.2 compatibility)
  };
}

// Poll a DAG run until terminal state
async function waitForDagRun(client, dagId, dagRunId, pollIntervalMs = 10_000, timeoutMs = 3_600_000) {
  const terminal = new Set(['success', 'failed']);
  const deadline = Date.now() + timeoutMs;

  while (Date.now() < deadline) {
    const run = await client.request('GET', `/dags/${dagId}/dagRuns/${dagRunId}`);

    if (terminal.has(run.state)) {
      return {
        dagRunId: run.dag_run_id,
        state: run.state,
        startDate: run.start_date,
        endDate: run.end_date,
        durationSec: run.end_date
          ? (new Date(run.end_date) - new Date(run.start_date)) / 1000
          : null,
      };
    }

    await new Promise((r) => setTimeout(r, pollIntervalMs));
  }

  throw new Error(`DAG run ${dagRunId} timed out after ${timeoutMs}ms`);
}

TaskInstance state machine and failure analysis

Task instances have a richer state space than DAG runs. The nine possible states reflect the complexity of retry logic, dependency resolution, and scheduling: none (initial state before scheduling), scheduled (ready for executor), queued (sent to executor, waiting for a worker), running (executing), success (completed), failed (failed and retry limit reached), upstream_failed (a dependency task failed — this task will not run), skipped (conditional branch or ShortCircuitOperator decided this task should not run), and up_for_retry (failed but retries remain). The up_for_reschedule state is specific to sensors: a sensor that hasn't found its target yet and is sleeping between pokes.

For failure analysis, the most useful combination is listing all TaskInstances for a failed DagRun filtered to failed and upstream_failed states, then fetching logs for the failed instances. The upstream_failed tasks are secondary failures — fix the root failed task and the upstream_failed ones will unblock on retry.

async function analyzeDagRunFailure(client, dagId, dagRunId) {
  // Get all task instances for this run
  const response = await client.request(
    'GET',
    `/dags/${dagId}/dagRuns/${dagRunId}/taskInstances`
  );

  const allTasks = response.task_instances;

  // Categorize by state
  const failed = allTasks.filter((t) => t.state === 'failed');
  const upstreamFailed = allTasks.filter((t) => t.state === 'upstream_failed');
  const skipped = allTasks.filter((t) => t.state === 'skipped');
  const successful = allTasks.filter((t) => t.state === 'success');

  // For each failed task, identify try number for log fetching
  const failedWithLogs = await Promise.all(
    failed.map(async (task) => {
      // try_number is the last attempt count (1-indexed)
      const tryNumber = task.try_number;
      const logUrl = `/dags/${dagId}/dagRuns/${dagRunId}/taskInstances/${task.task_id}/logs/${tryNumber}`;

      return {
        taskId: task.task_id,
        state: task.state,
        tryNumber,
        startDate: task.start_date,
        endDate: task.end_date,
        duration: task.duration,
        operator: task.operator,
        logUrl,
        maxTries: task.max_tries,
        retryExhausted: task.try_number > task.max_tries,
      };
    })
  );

  return {
    summary: {
      total: allTasks.length,
      success: successful.length,
      failed: failed.length,
      upstreamFailed: upstreamFailed.length,
      skipped: skipped.length,
    },
    rootCauses: failedWithLogs, // Start debugging here
    cascadeFailures: upstreamFailed.map((t) => t.task_id), // These will auto-fix when rootCauses are fixed
  };
}

// Fetch task execution log for a specific attempt
async function getTaskLog(client, dagId, dagRunId, taskId, tryNumber = 1) {
  // Airflow returns log as plain text with Accept: text/plain, or structured JSON
  const url = `${client.baseUrl}/api/v1/dags/${dagId}/dagRuns/${dagRunId}/taskInstances/${taskId}/logs/${tryNumber}`;
  const response = await fetch(url, {
    headers: { 'Authorization': client.authHeader, 'Accept': 'text/plain' },
  });

  if (!response.ok) throw new Error(`Log fetch failed: ${response.status}`);

  const logText = await response.text();
  // Trim to last N lines to avoid overwhelming LLM context
  const lines = logText.split('\n');
  const MAX_LINES = 200;
  return lines.length > MAX_LINES
    ? lines.slice(-MAX_LINES).join('\n')
    : logText;
}

DAG management: pause, unpause, and list recent runs

DAGs in Airflow have an is_paused flag that controls whether the scheduler creates new DagRuns for that DAG. A paused DAG will not trigger on its schedule but can still be triggered manually via the API. This is useful for deploying a DAG change that needs to be validated before the next scheduled run. MCP tools that manage Airflow pipelines often need to pause/unpause DAGs as part of deployment workflows.

Listing recent DAG runs is a common tool requirement for dashboards and health checks. The REST API supports pagination via offset and limit query parameters and filtering by state. To list only failed runs in the past 24 hours, combine state=failed with start_date_gte and start_date_lte filters.

async function pauseOrUnpauseDag(client, dagId, paused) {
  const dag = await client.request('PATCH', `/dags/${dagId}`, { is_paused: paused });
  return { dagId, isPaused: dag.is_paused };
}

async function getRecentFailedRuns(client, dagId, sinceHours = 24) {
  const since = new Date(Date.now() - sinceHours * 3_600_000).toISOString();
  const params = new URLSearchParams({
    state: 'failed',
    start_date_gte: since,
    limit: '100',
    order_by: '-start_date',
  });

  const response = await client.request('GET', `/dags/${dagId}/dagRuns?${params}`);

  return {
    totalCount: response.total_entries,
    runs: response.dag_runs.map((run) => ({
      dagRunId: run.dag_run_id,
      logicalDate: run.logical_date,
      startDate: run.start_date,
      endDate: run.end_date,
      state: run.state,
      externalTrigger: run.external_trigger, // true if triggered manually, false if scheduled
    })),
  };
}

// List all DAGs with their current pause and active states
async function listDags(client, onlyActive = true) {
  const params = new URLSearchParams({ limit: '200' });
  if (onlyActive) params.set('only_active', 'true');

  const response = await client.request('GET', `/dags?${params}`);
  return response.dags.map((dag) => ({
    dagId: dag.dag_id,
    isPaused: dag.is_paused,
    isActive: dag.is_active, // false if DAG file was removed but run history remains
    schedule: dag.schedule_interval, // null | '@daily' | '0 9 * * *'
    lastParsedTime: dag.last_parsed_time,
    fileloc: dag.fileloc, // absolute path to the DAG file on the scheduler
    tags: dag.tags?.map((t) => t.name) ?? [],
  }));
}

Scheduler health and AliveMCP integration

The GET /api/v1/health endpoint returns scheduler and metadata database health without requiring authentication. The scheduler health object contains latest_scheduler_heartbeat — the ISO timestamp of the last time the scheduler process wrote a heartbeat to the database. If this timestamp is more than 30 seconds old, the scheduler is likely down or stuck, and no new task slots are being filled even if the webserver is responding normally.

Register AliveMCP with a 60-second probe interval on your /health/airflow endpoint. The probe should check scheduler heartbeat freshness (alert if stale >60 seconds), metadata database connectivity, and optionally a count of currently running DAG runs (detect runaway pipelines). Do not rely solely on HTTP 200 from the webserver — a healthy webserver with a dead scheduler creates a deceptive false-positive.

async function checkAirflowHealth(client) {
  const health = await client.checkHealth();

  const schedulerHeartbeat = health.scheduler?.latest_scheduler_heartbeat;
  const heartbeatAge = schedulerHeartbeat
    ? Date.now() - new Date(schedulerHeartbeat).getTime()
    : Infinity;

  return {
    webserverReachable: true, // If we got here, webserver responded
    schedulerStatus: health.scheduler?.status ?? 'unknown',
    schedulerHeartbeatAge: heartbeatAge,
    schedulerHealthy: health.scheduler?.status === 'healthy' && heartbeatAge < 60_000,
    metadatabaseStatus: health.metadatabase?.status ?? 'unknown',
    metadatabaseHealthy: health.metadatabase?.status === 'healthy',
    // Overall health: all components must be healthy
    healthy: health.scheduler?.status === 'healthy'
      && health.metadatabase?.status === 'healthy'
      && heartbeatAge < 60_000,
  };
}

Related guides

• MCP tools for Temporal — workflow execution queries, signals, and namespace health
• MCP tools for AWS Step Functions — STANDARD vs EXPRESS, execution history, waitForTaskToken
• MCP tools for Prefect — flow run states, work pool health, and deployment triggers
• MCP tools for Dagster — asset materializations, run status, and sensor management
• MCP server health check patterns — composite endpoints and failure classification
• MCP server error handling — retries, timeouts, and structured error responses