Incident Response MCP for Apache Airflow

Apache Airflow MCP サーバー

独立したコミュニティプロジェクトです。Apache Software Foundation とは提携しておらず、同財団からの承認も受けていません。

PyPI、GHCR、公式 MCP Registry で公開されています。 再現可能なスキャンレポート、依存関係の監査結果、SBOM、来歴情報は、各不変リリースに添付されています。

Claude、Cursor、VS Code、またはその他のMCPクライアントを Apache Airflow のデプロイ環境に接続し、エージェントによる DAG の失敗原因の診断を支援できます。

PagerDuty／Datadog のアラートから Airflow UI のリンクを貼り付けて、「なぜ失敗したのか」と尋ねてみてください。エージェントは URL を解決し、失敗したタスクを特定して、MCP の応答前にフィルタリングとサイズ制限を適用したログエラーを取得し、実行の再トリガーやクリアも行えます。書き込みツールには破壊的操作を示す注釈が付与されており、MCP クライアントから確認を求める際に利用できます。

主な特長

• 🔍 インシデント対応を最優先 — Airflow UI の URL から失敗したタスクを直接特定し、コンテキスト行を含めてエラーレベルでログをフィルタリングし、try_number のセマンティクスを正しく扱います（センサーを含む）
• 🏢 複数インスタンス対応 — 同じ API ファミリーの開発／ステージング／本番環境を 1 つのサーバーで扱えます。インスタンスごとの認証情報に対応し、未知のホストを拒否する SSRF ガードを備えています
• 🔒 安全制御 — オプトインの読み取り専用モード（AIRFLOW_MCP_READ_ONLY=true）では書き込みツールを登録しません。書き込みツールには破壊的操作の注釈が付与され、設定済みの認証情報はインスタンスの応答や操作ログから秘匿されます
• 📉 トークン効率 — LLM のコンテキストウィンドウを考慮して、ログ末尾の取得、レベル別フィルタリング、バイト数の上限、切り詰め情報を設計しています
• 🧭 Airflow 2 と 3 に対応 — API v1／v2 アダプターを備え、JWT 認証を含めて Airflow 2.11 および 3.3 とのライブ E2E テストを実施しています
• 📎 追跡可能 — すべての応答に、構造化サーバーログと照合できる request_id が含まれます

クイックスタート

1. インストール

uv tool install apache-airflow-mcp-server \
  --with 'apache-airflow-client==3.3.0'  # replace 3.3.0 with your Airflow version

Airflow 2.11 をお使いですか？ 代わりに --with 'apache-airflow-client==2.10.0' を使用してください。2.10.0 は生成された v1 クライアントの最終版で、Airflow 2 の安定版 API を対象としています。別のリリースを使用する前に、Airflow との互換性を確認してください。

2. MCP クライアントを接続する

最も手早い方法は、環境変数だけで設定する単一インスタンスです。設定ファイルは必要ありません。

claude mcp add airflow \
  --env AIRFLOW_MCP_HOST=https://airflow.example.com \
  --env AIRFLOW_MCP_USERNAME=admin \
  --env AIRFLOW_MCP_PASSWORD=your-password \
  --env AIRFLOW_MCP_READ_ONLY=true \
  -- uvx --from apache-airflow-mcp-server \
  --with apache-airflow-client==3.3.0 airflow-mcp --transport stdio

{
  "mcpServers": {
    "airflow": {
      "command": "uvx",
      "args": ["--from", "apache-airflow-mcp-server", "--with", "apache-airflow-client==3.3.0", "airflow-mcp", "--transport", "stdio"],
      "env": {
        "AIRFLOW_MCP_HOST": "https://airflow.example.com",
        "AIRFLOW_MCP_USERNAME": "admin",
        "AIRFLOW_MCP_PASSWORD": "your-password",
        "AIRFLOW_MCP_READ_ONLY": "true"
      }
    }
  }
}

{
  "mcpServers": {
    "airflow": {
      "command": "uvx",
      "args": ["--from", "apache-airflow-mcp-server", "--with", "apache-airflow-client==3.3.0", "airflow-mcp", "--transport", "stdio"],
      "env": {
        "AIRFLOW_MCP_HOST": "https://airflow.example.com",
        "AIRFLOW_MCP_USERNAME": "admin",
        "AIRFLOW_MCP_PASSWORD": "your-password",
        "AIRFLOW_MCP_READ_ONLY": "true"
      }
    }
  }
}

{
  "servers": {
    "airflow": {
      "type": "stdio",
      "command": "uvx",
      "args": ["--from", "apache-airflow-mcp-server", "--with", "apache-airflow-client==3.3.0", "airflow-mcp", "--transport", "stdio"],
      "env": {
        "AIRFLOW_MCP_HOST": "https://airflow.example.com",
        "AIRFLOW_MCP_USERNAME": "admin",
        "AIRFLOW_MCP_PASSWORD": "your-password",
        "AIRFLOW_MCP_READ_ONLY": "true"
      }
    }
  }
}

AIRFLOW_MCP_HOST=https://airflow.example.com \
AIRFLOW_MCP_USERNAME=admin AIRFLOW_MCP_PASSWORD=your-password \
AIRFLOW_MCP_READ_ONLY=true \
airflow-mcp --transport http --host 127.0.0.1 --port 8765

{ "mcpServers": { "airflow": { "url": "http://127.0.0.1:8765/mcp" } } }

3. エージェントに質問する

「etl_pipeline の最新の実行が失敗したのはなぜですか？」

「https://airflow.example.com/dags/etl_pipeline/grid — ここで何が起きたのでしょうか？クリアしても安全ですか？」

「失敗したタスクのエラーコンテキストを表示し、最小限の復旧アクションを教えてください。」

Airflow との互換性

このサーバーは、生成された apache-airflow-client を通じて Airflow と通信します。 Airflow 3 では、クライアントのリリースを Airflow のリリースに合わせてください。生成モデルはメジャーバージョン内でも変更される可能性があり、新しいクライアントが古いサーバーのレスポンスを正しくデシリアライズできるとは限りません。Airflow 2.11 では、Airflow 2 の安定版 API に対して、最後の v1 クライアントリリースである 2.10.0 を使用します。

api_version が設定されていない場合、サーバーはインストール済みクライアントに対応する API を前提とします（2.x クライアントでは v1、3.x クライアントでは v2）。メジャーバージョンの不一致を早期に検出するには、AIRFLOW_MCP_API_VERSION（またはレジストリ内の api_version:）を明示的に設定してください。

現在、1 つのサーバープロセスで読み込める生成クライアントは 1 つのメジャーバージョンのみです。そのため、レジストリ内のすべてのインスタンスで同じ API ファミリーを使用する必要があります。Airflow 2 と Airflow 3 には、別々の MCP サーバープロセスを実行してください。混在バージョンのサポートには今後のクライアントアダプターの変更が必要で、現時点では動作するものとして案内していません。

Airflow 3 に関する注意事項：

• 認証：ベアラートークンは JWT として引き渡されます。基本認証情報は POST /auth/token 経由で自動的に JWT と交換され、定期的に更新されます（AIRFLOW_MCP_TOKEN_REFRESH_SECONDS のデフォルトは 3600 です。デプロイ環境の JWT 有効期限より短く設定してください。また、現時点では 401 時に自動で再認証されないことに注意してください）。
• execution_date の順序は logical_date に対応し、データセットはアセットに対応し、UI リンクには Airflow 3 のルート方式が使用されます。ツール名と主要なワークフローは安定していますが、ドキュメントに記載されているフィールドやオプションは API ファミリーによって異なる場合があります。
• Airflow 3 で存在しなくなったクリアオプション（include_subdags／include_parentdag、および airflow_clear_dag_run の include_*／reset_dag_runs オプション）は、破壊的な操作の範囲を暗黙に狭めるのではなく、INVALID_INPUT として拒否されます。

両方のクライアントファミリーは、該当するプルリクエストと main へのプッシュでテストされています。実際の Airflow デプロイ環境からのバグ報告を歓迎します！

設定

単一インスタンス（環境変数のみ）

* ユーザー名とパスワード、またはトークンのいずれかを指定してください。

複数インスタンス（レジストリ YAML）

AIRFLOW_MCP_INSTANCES_FILE に YAML レジストリを指定します（単一インスタンス用の環境変数より優先されます）。値では ${VAR} を使って環境変数を参照できます。

data-stg:
  host: https://airflow.data-stg.example.com/
  api_version: v1        # Airflow 2
  verify_ssl: true
  auth:
    type: basic
    username: ${AIRFLOW_DATA_STG_USERNAME}
    password: ${AIRFLOW_DATA_STG_PASSWORD}

data-prod:
  host: https://airflow.data-prod.example.com/
  api_version: v1        # Keep one client/API family per server process
  auth:
    type: bearer
    token: ${AIRFLOW_DATA_PROD_TOKEN}

すべてのツールでは、instance キー（data-stg）または ui_url（ホストがレジストリと照合される完全な http(s) Airflow UI URL）を指定できます。未知のホストは拒否されます（SSRF ガード）。リンクに dag_id／dag_run_id／task_id が含まれる場合、ui_url によってそれらも自動入力されます。instance と ui_url の両方が渡され、指定内容が一致しない場合は、推測せずに INSTANCE_MISMATCH で呼び出しが失敗します。

Kubernetes のヒント： レジストリを Secret から /config/instances.yaml にマウントし、AIRFLOW_MCP_INSTANCES_FILE=/config/instances.yaml を設定します。

サーバーオプション

読み取り専用モード

クイックスタートでは AIRFLOW_MCP_READ_ONLY=true を設定しています。書き込みツール（トリガー、クリア、一時停止／一時停止解除）は登録されません。これにより MCP による変更を防げますが、読み取りツールから機密性の高いログ、設定、レンダリング済みフィールド、DAG 実行データが開示される可能性はあります。最小権限の Airflow 認証情報を使用してください。

リカバリー操作を意図的に有効にするには、AIRFLOW_MCP_READ_ONLY=false を設定します。その場合、書き込みツールには MCP の destructiveHint アノテーションが付与され、クライアントは確認を求めるかどうかの判断に利用できます。アノテーションはあくまで助言であるため、対象環境に適した Airflow 認証情報と MCP クライアントの承認動作が整っていない限り、書き込みを有効にしないでください。

ツール

検出と URL ユーティリティ

読み取り

書き込み（破壊的操作として注釈されているため、クライアント側で承認を必須にできます。読み取り専用モードでは完全に非表示になります）

成功時のペイロードにはすべてログとの照合に使える request_id が含まれます。失敗時には、{code, message, request_id, context} を含む構造化された ToolError が発生します。

インシデント対応ワークフロー

このツールの設計で想定しているフローです。アラートリンクから診断まで、4 回の呼び出しで進めます。

# 1. Alert contains an Airflow UI link → resolve it
airflow_resolve_url("https://airflow.example.com/dags/etl_pipeline/grid?dag_run_id=...")
#    → {instance, dag_id, dag_run_id, ...}

# 2. Which tasks failed in this run?
airflow_list_task_instances(dag_id="etl_pipeline", dag_run_id="scheduled__2026-01-01",
                            state=["failed"])

# 3. Get attempt metadata (authoritative try_number, retries, timings)
ti = airflow_get_task_instance(dag_id="etl_pipeline",
                               dag_run_id="scheduled__2026-01-01",
                               task_id="transform_data")

# 4. Pull only the error lines, with context, capped for the LLM
airflow_get_task_instance_logs(dag_id="etl_pipeline",
                               dag_run_id="scheduled__2026-01-01",
                               task_id="transform_data",
                               try_number=ti["attempts"]["try_number"],
                               tail_lines=500, filter_level="error", context_lines=5)

ログのレスポンスには truncated、auto_tailed（100 MB を超えるログは自動的に末尾を取得）、match_count、バイト数／行数の統計情報が含まれるため、エージェントは確認対象を正確に把握できます。ホストごとに分割されたログは --- [worker-1] --- ヘッダー付きでフラット化されます。Airflow 3 の構造化ログは通常の行として出力されます。

try_number に関する注意： 再スケジュールモードのセンサーでは、Airflow 2.9 までは再スケジュールのたびに値が増えることがありました。Airflow 2.10 以降では増えなくなっています。推測せず、必ず airflow_get_task_instance から読み取ってください。派生フィールドの retries_consumed／retries_remaining はヒューリスティックです。

デプロイ

Docker

docker run -p 127.0.0.1:8765:8765 \
  -e AIRFLOW_MCP_HOST=https://airflow.example.com \
  -e AIRFLOW_MCP_USERNAME=admin \
  -e AIRFLOW_MCP_PASSWORD=your-password \
  -e AIRFLOW_MCP_READ_ONLY=true \
  ghcr.io/madamak/apache-airflow-mcp-server:latest

または、docker build -t airflow-mcp . でローカルにビルドできます。

リリースイメージにはロックファイルで固定された Airflow 3.3 クライアントが含まれ、:8765 でストリーム可能な HTTP を提供します（/mcp エンドポイント、プローブ用の /health）。MCP HTTP エンドポイントには呼び出し元の認証機能が組み込まれていません。ループバックにバインドするか、認証済みのプライベートプロキシの背後に配置してください。複数インスタンスのセットアップでは、同じ API ファミリーのレジストリ YAML をマウントしてください。別個の v1 イメージが公開されるまでは、Airflow 2 のデプロイには上記の固定済みローカルインストール方法を使用してください。

CI では、イメージにインストールされた Python 依存関係を正確に監査し、固定された Cisco MCP Scanner リリースの YARA アナライザーを使って、読み取り専用と書き込み可能の両方の MCP ツールサーフェスをスキャンします。スキャンが不完全な場合、または未トリアージの YARA 検出がある場合、セキュリティワークフローは失敗します。v1.0.1 以降、リリースアセットには機械可読なスキャンレポートが含まれ、リリースイメージのマニフェストには、より広範なパッケージ一覧を対象とする SBOM と来歴の証明が添付されます。これらは自動チェックであり、セキュリティ認証でも、デプロイ固有のレビューの代替でもありません。

FastMCP ツール

FastMCP 対応ツールがエントリーポイントとデプロイのデフォルト設定を自動検出できるよう、fastmcp.json が含まれています。

開発

uv sync                 # install dependencies
uv run pytest           # unit tests (no real network; the Airflow client is mocked)
uv run ruff check .     # lint
uv run ruff format .    # format
uv run airflow-mcp --transport stdio   # run locally
./scripts/e2e.sh af2    # end-to-end against Airflow 2.11
./scripts/e2e.sh af3    # end-to-end against Airflow 3.3
                        # Both seed failures/noisy logs and drive every tool
                        # through MCP. Set E2E_KEEP=1 to keep the instance up.

CI では、Python 3.10～3.13 で apache-airflow-client の両ファミリーに対してユニットテストを実行します。該当するプルリクエスト、main へのプッシュ、夜間実行、リリースでは、Docker 化した Airflow 2.11 と 3.3 に対する実動テストも行います。ガイドラインについては CONTRIBUTING.md を、コーディングエージェントにこのリポジトリを参照させる場合は AGENTS.md をご覧ください（その用途向けに書かれています）。

コントリビューション

Issue や PR を歓迎します。特に、次のような内容をお待ちしています。

• 実際の Airflow のインシデント対応ワークフローやバージョンの組み合わせに関する報告
• ログの範囲制限、診断の安全性、URL を優先するワークフローの改善
• より多くの MCP ホストやデプロイの種類に対応するクライアント設定手順

このサーバーによってデバッグに費やす時間を減らせたなら、⭐ を付けていただくと、他の Airflow チームにも見つけてもらいやすくなります。

ライセンス

Apache 2.0 — LICENSE を参照してください。

Apache Airflow および Airflow は、The Apache Software Foundation の登録商標です。この独立したプロジェクトは ASF と提携しておらず、ASF の承認も受けていません。