MCP Server Apache Airflow

Apache Airflow用に実装されたMCPプロトコルサーバーで、標準化されたインターフェースを通じてAirflowとやり取りできます。

開発者ツールクラウドプラットフォーム #ワークフロー管理 #APIゲートウェイ #Airflow統合 #MCPプロトコル .Python

スコア : 2.5ポイント

ダウンロード数 : 7.8K

更新時間 : 2025-04-28

サイトを開く

Apache Airflow MCPサーバーとは？

Apache Airflow MCPサーバーは、標準MCPプロトコルに基づくツールで、Apache Airflowの強力な機能を統一されたAPIインターフェースにカプセル化し、ユーザーが簡単なコマンドとインターフェースでAirflowタスク、DAG、および変数を簡単に管理できるようにします。

Apache Airflow MCPサーバーの使い方は？

ユーザーは環境変数を設定してサーバーを起動するだけで、標準MCPクライアントまたはコマンドラインツールを通じてAirflowとやり取りできます。

適用シナリオ

複雑なワークフローの自動管理と監視が必要なチームに適しており、特にデータ処理、機械学習、およびビジネスプロセス自動化の分野で最適です。

主な機能

DAG管理

DAGの作成、削除、一時停止、再開、および一括更新などの機能をサポートします。

DAG実行管理

DAG実行インスタンスを表示、作成、変更、および削除できます。

タスクインスタンス管理

タスクインスタンスの状態を操作します。例えば、クリア、状態設定などです。

変数管理

変数の追加、削除、変更、および検索操作をサポートします。

データセット管理

データセットイベントと依存関係を追跡および管理します。

ヘルスモニタリング

システムの実行状態とバージョン情報を取得します。

利点

Airflowの複雑なAPIの使用体験を簡素化します。

複数のクライアントとのシームレスな統合をサポートします。

強力なエラー処理メカニズムで安定性を確保します。

制限

正常に動作するには環境変数を正しく設定する必要があります。

一部の高度な機能には追加の設定が必要な場合があります。

使い方

インストールと設定

サーバーの依存関係をインストールし、必要な環境変数（AIRFLOW_HOST、AIRFLOW_USERNAME、AIRFLOW_PASSWORDなど）を設定します。

サーバーの起動

サーバーを実行してMCPリクエストを待機します。

接続のテスト

サーバーが正常に起動し、リクエストに応答することを確認します。

使用例

DAG実行のトリガー

APIを通じて新しいDAG実行をトリガーします。

DAG設定の更新

既存のDAGの設定パラメータを変更します。

よくある質問

環境変数を設定するにはどうすればいいですか？

権限問題が発生した場合はどうすればいいですか？

どのオペレーティングシステムがサポートされていますか？

🚀 Apache Airflow MCP サーバ

MCP（Maintenance Command Processor）は、Apache Airflow と統合されたサードパーティサービスを管理するためのツールです。このツールを使用することで、様々なメンテナンスコマンドを実行し、関連するタスクを処理することができます。

✨ 主な機能

概要

MCP（Maintenance Command Processor）は、Apache Airflow と統合されたサードパーティサービスを管理するためのツールです。MCP を通じて、様々なメンテナンスコマンドを実行し、関連するタスクを処理することができます。

サポートされる API グループ

以下は、MCP サーバがサポートするすべての API グループとそれに対応するインターフェースです。

グループ	インターフェース
config	GET /api/v1/config
connections	GET, DELETE /api/v1/connections/{connection_id}
dag	GET, POST /api/v1/dags, GET, DELETE /api/v1/dags/{dag_id}, PATCH /api/v1/dags/{dag_id}/paused
dagrun	GET, POST /api/v1/dagruns, GET /api/v1/dagruns/{dag_run_id}
dagstats	GET /api/v1/dags/statistics
dataset	GET, DELETE /api/v1/datasets/{dataset_uri}, GET, DELETE /api/v1/datasets/{dataset_uri}/dag_runs/queued_events
eventlog	GET, DELETE /api/v1/event_logs/{event_log_id}
importerror	GET, DELETE /api/v1/import_errors/{import_error_id}
monitoring	GET /api/v1/health
plugin	GET /api/v1/plugins
provider	GET /api/v1/providers
taskinstance	GET, PATCH, DELETE /api/v1/task_instances/{task_instance_id}, GET /api/v1/dags/{dag_id}/task_instances
variable	GET, DELETE /api/v1/variables/{variable_key}
xcom	GET, DELETE /api/v1/xcoms/{xcom_key}

📦 インストール

依存関係

このプロジェクトは、公式の Apache Airflow クライアントライブラリ (apache-airflow-client) に依存しており、インストール時に自動的に処理されます。

環境変数

以下の環境変数を設定してください。

AIRFLOW_HOST=<あなたの Airflow ホスト>
AIRFLOW_USERNAME=<あなたの Airflow ユーザー名>
AIRFLOW_PASSWORD=<あなたの Airflow パスワード>

Claude Desktop との統合

claude_desktop_config.json ファイルに以下の内容を追加してください。

{
  "mcpServers": {
    "mcp-server-apache-airflow": {
      "command": "uvx",
      "args": ["mcp-server-apache-airflow"],
      "env": {
        "AIRFLOW_HOST": "https://あなたの Airflow ホスト",
        "AIRFLOW_USERNAME": "あなたのユーザー名",
        "AIRFLOW_PASSWORD": "あなたのパスワード"
      }
    }
  }
}

uv を使用する代替構成は以下の通りです。

{
  "mcpServers": {
    "mcp-server-apache-airflow": {
      "command": "uv",
      "args": [
        "--directory",
        "/実際のパス/mcp-server-apache-airflow",
        "run",
        "mcp-server-apache-airflow"
      ],
      "env": {
        "AIRFLOW_HOST": "https://あなたの Airflow ホスト",
        "AIRFLOW_USERNAME": "あなたのユーザー名",
        "AIRFLOW_PASSWORD": "あなたのパスワード"
      }
    }
  }
}

/実際のパス/mcp-server-apache-airflow を、あなたが実際にコードをクローンした場所に置き換えてください。

カスタム API グループ

--apis フラグを使用して、使用する API グループを選択します。

# すべてのデフォルト API グループを有効にする
$ mcp-server-apache-airflow --apis default

# 特定の API グループを有効にする（例: config, dag）
$ mcp-server-apache-airflow --apis config,dag

💻 使用例

環境変数の設定

以下の環境変数を設定してください。

AIRFLOW_HOST: Airflow サーバーのアドレス。デフォルトは http://localhost:8080 です。
AIRFLOW_USERNAME: ログインするユーザー名。デフォルトは admin です。
AIRFLOW_PASSWORD: ユーザーのパスワード。デフォルトは <PASSWORD> です。

MCP サービスの起動

プロジェクトのルートディレクトリで以下のコマンドを実行して、MCP サーバーを起動します。

$ python -m mcp_server --apis default

🔧 技術詳細

API エンドポイント

設定インターフェース

設定の取得
```
GET /api/v1/config
```

接続インターフェース

すべての接続の一覧表示
```
GET /api/v1/connections
```

指定された接続の削除

DELETE /api/v1/connections/{connection_id}

DAG インターフェース

DAG の作成または一覧表示
```
POST, GET /api/v1/dags
```
指定された DAG の情報の取得
```
GET /api/v1/dags/{dag_id}
```
指定された DAG の削除
```
DELETE /api/v1/dags/{dag_id}
```
DAG の一時停止または再開
```
PATCH /api/v1/dags/{dag_id}/paused
```

DAG 実行インターフェース

DAG 実行の作成または一覧表示
```
POST, GET /api/v1/dagruns
```
指定された DAG 実行の情報の取得
```
GET /api/v1/dagruns/{dag_run_id}
```

データセットインターフェース

データセットの作成または一覧表示
```
POST, GET /api/v1/datasets
```
指定されたデータセットの削除
```
DELETE /api/v1/datasets/{dataset_uri}
```
データセットに関連付けられたキューイベントの取得
```
GET /api/v1/datasets/{dataset_uri}/dag_runs/queued_events
```
データセットに関連付けられたキューイベントの削除
```
DELETE /api/v1/datasets/{dataset_uri}/dag_runs/queued_events
```

イベントログインターフェース

指定されたイベントログの取得
```
GET /api/v1/event_logs/{event_log_id}
```

指定されたイベントログの削除

DELETE /api/v1/event_logs/{event_log_id}

インポートエラーインターフェース

指定されたインポートエラーの取得
```
GET /api/v1/import_errors/{import_error_id}
```

指定されたインポートエラーの削除

DELETE /api/v1/import_errors/{import_error_id}

監視インターフェース

システムのヘルス状態の取得
```
GET /api/v1/health
```

プラグインインターフェース

すべてのプラグインの一覧表示
```
GET /api/v1/plugins
```

プロバイダーインターフェース

すべてのプロバイダーの一覧表示
```
GET /api/v1/providers
```

タスクインスタンスインターフェース

タスクインスタンスの作成または一覧表示
```
POST, GET /api/v1/task_instances
```
指定されたタスクインスタンスの情報の取得
```
GET /api/v1/task_instances/{task_instance_id}
```
指定されたタスクインスタンスの更新または削除
```
PATCH, DELETE /api/v1/task_instances/{task_instance_id}
```
DAG 内のタスクインスタンスの一覧表示
```
GET /api/v1/dags/{dag_id}/task_instances
```