Airflow MCPサーバーでDAGの実行状況をClaudeに監視させる

Airflow MCPサーバーとは何か

Airflow MCPサーバー(astro-airflow-mcp)は、Apache AirflowのREST APIをMCPツールとして公開するサーバーです。提供元はAirflowの商用ベンダーであるAstronomerで、Apache Airflowプロジェクト本体のものではありません。GitHubではAstronomerのagentsリポジトリ配下のastro-airflow-mcpディレクトリで公開されています。

FastMCPで実装され、DAG・タスク・プール・変数・接続・アセットといったAirflowの主要リソースをツール化しています。用途は大きく2つで、DAGの実行状況を自然言語で照会する運用監視と、失敗したタスクインスタンスのログを辿る原因調査です。ライセンスはApache 2.0、PyPIの最新版は0.9.1、Pythonは3.10以上が必要です。

Claude Codeに接続する

インストール不要で、uvxがPyPIから直接実行します。

claude mcp add airflow -- uvx astro-airflow-mcp --transport stdio

接続先の既定値はhttp://localhost:8080で、Astro CLIで立ち上げたローカル環境がそのまま対象になります。リモートのAirflowに繋ぐときは、接続先と認証情報を-eで渡します。DAGを書く側の検証はClaude CodeでAirflowのDAGを書く手順にあります。

claude mcp add airflow \
  -e AIRFLOW_API_URL=https://your-airflow.example.com \
  -e AIRFLOW_USERNAME=admin \
  -e AIRFLOW_PASSWORD=admin \
  -- uvx astro-airflow-mcp --transport stdio

渡せる環境変数は次のとおりです。書き込みを止めるAF_READ_ONLYと、社内CAのための2つが、運用に入ると効いてきます。

--transportの既定値は、READMEの中で食い違っています。注記には「既定はHTTPモードなので--transport stdioが必須」とあり、スタンドアロンの起動例にも「HTTP mode is default」とあります。一方、同じREADMEのオプション表は既定値をstdioとしています。どちらが実際の挙動でも困らないように、claude mcp addでは常に--transport stdioを付けておきます。

認証はAirflowのバージョンで挙動が変わる

Airflow 2.xと3.xでは認証APIが異なり、astro-airflow-mcpはアダプターパターンで両方を自動判定します。Airflow 2.xはBasic認証(/api/v1)、Airflow 3.xはユーザー名とパスワードからOAuth2トークンを交換する方式(/api/v2)です。AIRFLOW_USERNAMEとAIRFLOW_PASSWORDを渡せば、どちらの方式かはサーバー側が切り替えます。

固定のBearerトークンなら、AIRFLOW_AUTH_TOKENが両バージョンで使えます。起動時にAPIエンドポイントへプローブを送ってバージョンを検出するので、接続先が2.xか3.xかを事前に調べる必要はありません。

使えるツールの分類

ツールは、エージェント向けに複数の情報をまとめて返す統合ツールと、個別リソースを扱うコアツールの2層構成です。

list_connectionsは認証情報を除いて返すため、接続一覧をClaudeに見せても資格情報はやり取りに乗りません。ログが長大になりがちなタスクを調べるときは、MAX_MCP_OUTPUT_TOKENSでMCPツール出力の上限を調整すると読み込みが安定します。

プール(list_pools)と変数(list_variables)は、DAGの並列実行数やDAG間で共有する設定値を管理するAirflow標準の仕組みです。get_poolは個別のプールの詳細を取得するツールで、etl_poolのようにプールを指定して聞けば、そのプールの詳細を確かめられます。アセット系のlist_assetsは、Airflow 2.xのデータセットと3.xのアセットを同じ名前で扱えるようにしたもので、DAG間のデータ依存を辿るときに使います。

ツールのほかに、MCPのリソースとプロンプトも公開されています。リソースはairflow://version・airflow://providers・airflow://plugins・airflow://configの4つです。プロンプトは次の3つで、Claude Codeでは/mcp__airflow__に続けて名前を打つとコマンドとして実行できます(サーバー名をairflowで登録した場合)。

失敗したDAGの原因を自然言語で調べる

症状から入るなら、統合ツールに寄せた聞き方がそのまま効きます。調査は、システム全体から個別のrunへと絞る流れが基本です。

get_task_instanceは実行結果を読むだけのツールで、再実行はしません。この区別は、読み取り専用の運用を組むときにも効いてきます。

書き込みを止める3つの層

Airflowの読み取りだけを任せたい場合、止める場所は3つあります。層ごとに効く範囲が違うので、重ねて使えます。

Claude Code側の設定は、プロジェクトの.claude/settings.jsonに次のように書きます。

{
  "permissions": {
    "deny": [
      "mcp__airflow__trigger_dag",
      "mcp__airflow__trigger_dag_and_wait",
      "mcp__airflow__pause_dag",
      "mcp__airflow__delete_dag_run",
      "mcp__airflow__clear_dag_run",
      "mcp__airflow__clear_task_instances"
    ]
  }
}

本番のスケジューラを動かす操作は、Kubernetes MCPサーバーのような他のインフラ系MCPと同じく、対話セッションで承認プロンプトを挟んで実行する形が無難です。

スタンドアロンとプラグインモードのどちらを選ぶか

ローカルの1人利用なら、ここまでのスタンドアロン起動で足ります。複数人で共有したい、リモートのAirflowに常設したいといった場合は、プラグインモードが選択肢になります。

プラグインモードのエンドポイントはhttps://<your-airflow-url>/mcp/v1/で、認証はAuthorizationヘッダーで渡します。Claude Codeには-t httpで登録します(-t sseではありません)。

claude mcp add -t http \
  -H "Authorization: Bearer <TOKEN>" \
  airflow "https://<your-airflow-url>/mcp/v1/"

Astronomer Astroのデプロイメントでは、Bearerトークンにデプロイメントのトークンかセッショントークンを使います。オープンソースのAirflow 2.xでは、構成した認証バックエンドに応じてBasic認証かBearerトークンです。Airflow 2.xではFlaskのブループリントとして登録されるため、各gunicornワーカーの最初の呼び出しだけ、わずかに遅れます。Airflow 3.xでは複数のAPIサーバーのレプリカがあってもセッションの固定が要りません。

プロバイダーとオペレーターの仕様を引く

afには、稼働中のAirflowに接続しなくても使えるaf registry系のコマンドがあります。MCPのツールではなくCLIなので、先にuv tool install astro-airflow-mcpでafを入れておき、Claude CodeにはBashで呼び出させます。

af registry providers
af registry modules amazon | jq '.modules[] | select(.type == "hook") | .name'
af registry parameters ftp

新しいDAGを書く前に「amazonプロバイダーのフックを一覧にして」と頼めば、一覧に加えて、af registry parametersでコンストラクターの引数も確認できます。結果は~/.af/.registry_cache/にキャッシュされ、バージョンを付けない問い合わせは1時間、--versionで固定した問い合わせは30日保持されます。最新を取り直したいときは--no-cacheを付けます。

af CLIでターミナルから直接操作する

astro-airflow-mcpパッケージには、MCPサーバーとは別にafというCLIが同梱されています。af dags listやaf runs list --dag-id <dag_id>でAirflowのREST APIを直接叩け、instance系のコマンドを除いて出力はJSONなのでjqと組み合わせられます。

af health
af dags explore daily_sales_etl
af runs list --dag-id daily_sales_etl | jq '.dag_runs[] | select(.state == "failed")'

MCP経由の自然言語操作と、afによるスクリプト向けの直接操作を使い分けられます。任意のエンドポイントを叩くaf apiもあり、af api lsで使えるエンドポイントを一覧できます。

afは複数のAirflowインスタンスをaf instance addで登録し、af instance use prodで切り替えます。設定ファイルは3つの範囲に分かれます。

共有ファイルは既定でコミットされるため、トークンを直書きせず${AIRFLOW_PROD_TOKEN}のような環境変数の展開で参照します。生の資格情報を置きたいときは、ローカルかグローバルに書きます。af instance useの切り替え先は、開発者ごとにローカルファイルへ記録されます。

導入でつまずきやすいポイント

旧リポジトリのastronomer/astro-airflow-mcpはアーカイブされて読み取り専用のため、新しいIssueは受け付けられません。現在の開発はastronomer/agents配下のastro-airflow-mcpディレクトリで行われています。

afは、接続先のインスタンスが未設定だとhttp://localhost:8080にフォールバックします。CIのように「Airflowが無い」ことを明示したい場面では、AIRFLOW_API_URL=""と空文字で設定します。READMEの例では、次のエラーで止まります。

AIRFLOW_API_URL="" af dags list
# Error: AIRFLOW_API_URL is set but empty. No Airflow instance is configured.

社内の自己署名証明書やプライベートCAを使うAirflowには、サーバー側のオプションで対応できます。--no-verify-ssl(環境変数はAIRFLOW_VERIFY_SSL=false)で検証を無効にするか、--ca-cert(環境変数はAIRFLOW_CA_CERT)でCAバンドルを指定します。af側ではaf instance addの--no-verify-sslと--ca-certが同じ役割です。

afは、使ったコマンド名だけを匿名のテレメトリーとして収集します。引数や値は含まれません。止めるにはaf telemetry disableを実行するか、環境変数AF_TELEMETRY_DISABLED=1を設定します。

まとめ

Airflow MCPサーバーは、DAGの実行状況の照会と失敗原因の調査を自然言語に落とし込めるMCP実装です。最初の一歩はget_system_healthとdiagnose_dag_runで、監視だけならAF_READ_ONLY=trueで書き込みを丸ごと止められます。再実行まで任せるなら、denyルールで危険なツールだけを個別に塞ぎ、残りは対話セッションで承認しながら進めます。

よくある質問

Airflow 2.xと3.xで使えるツールは同じか

コアツールの大半は両バージョンで共通ですが、DAG実行統計を返すget_dag_statsのようにAirflow 3.x限定のツールもあります。