MCP Server Apache Airflow

🚀 Apache Airflow MCP 服務器

MCP 服務器是 Apache Airflow 的實用工具，用於管理與 Airflow 集成的第三方服務。藉助它，用戶能夠執行各類維護命令並處理相關任務。

✨ 主要特性

簡介

MCP（Maintenance Command Processor）作為 Apache Airflow 的工具，專門用於管理與 Airflow 集成的第三方服務。通過 MCP，您可以輕鬆執行各種維護命令並處理相關任務。

支持的 API 組

以下是 MCP 服務器支持的所有 API 組及其對應的接口：

📦 安裝指南

依賴項

該項目依賴於官方 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 接口

• 創建或列出 DAGs

  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
  

變量接口

• 創建或列出變量

  POST, GET /api/v1/variables
  

• 獲取指定變量的值

  GET /api/v1/variables/{variable_key}
  

• 刪除指定變量

  DELETE /api/v1/variables/{variable_key}
  

XCom 接口

• 創建或列出 XCom 數據

  POST, GET /api/v1/xcoms
  

• 獲取指定 XCom 的值

  GET /api/v1/xcoms/{xcom_key}
  

• 刪除指定 XCom

  DELETE /api/v1/xcoms/{xcom_key}
  

📚 詳細文檔

貢獻指南

如何貢獻代碼

1. Fork 本倉庫
2. 創建功能分支
3. 提交代碼並創建 Pull Request

如何報告問題

請在 Issues 中提交您遇到的問題。

許可證

[此處添加許可證信息]