MCP服务器-Apache-Airflow

Name: MCP服务器-Apache-Airflow
Author: yangkyeongmo

作者：yangkyeongmo113 星标更新：2025-10-29

安装使用来源：spider==github.com

项目介绍

技术文档摘要

mcp-server-apache-airflow

PyPI - 下载量

这是一个用于 Apache Airflow 的 Model Context Protocol (MCP) 服务器实现，它能够与 MCP 客户端无缝集成。该项目提供了一种标准化的方式，通过 Model Context Protocol 与 Apache Airflow 进行交互。

关于

本项目实现了 Model Context Protocol，该协议封装了 Apache Airflow 的 REST API，允许 MCP 客户端以标准化方式与 Airflow 进行交互。它使用官方的 Apache Airflow 客户端库来确保兼容性和可维护性。

功能实现状态

功能	API 路径	状态
DAG 管理
列出 DAGs	`/api/v1/dags`	✅
获取 DAG 细节	`/api/v1/dags/{dag_id}`	✅
暂停 DAG	`/api/v1/dags/{dag_id}`	✅
恢复 DAG	`/api/v1/dags/{dag_id}`	✅
更新 DAG	`/api/v1/dags/{dag_id}`	✅
删除 DAG	`/api/v1/dags/{dag_id}`	✅
获取 DAG 源代码	`/api/v1/dagSources/{file_token}`	✅
批量更新多个 DAGs	`/api/v1/dags`	✅
解析 DAG 文件	`/api/v1/dagSources/{file_token}/reparse`	✅
DAG 运行
列出 DAG 运行	`/api/v1/dags/{dag_id}/dagRuns`	✅
创建 DAG 运行	`/api/v1/dags/{dag_id}/dagRuns`	✅
获取 DAG 运行细节	`/api/v1/dags/{dag_id}/dagRuns/{dag_run_id}`	✅
更新 DAG 运行	`/api/v1/dags/{dag_id}/dagRuns/{dag_run_id}`	✅
删除 DAG 运行	`/api/v1/dags/{dag_id}/dagRuns/{dag_run_id}`	✅
批量获取 DAG 运行	`/api/v1/dags/~/dagRuns/list`	✅
清除 DAG 运行	`/api/v1/dags/{dag_id}/dagRuns/{dag_run_id}/clear`	✅
设置 DAG 运行备注	`/api/v1/dags/{dag_id}/dagRuns/{dag_run_id}/setNote`	✅
获取上游数据集事件	`/api/v1/dags/{dag_id}/dagRuns/{dag_run_id}/upstreamDatasetEvents`	✅
任务
列出 DAG 任务	`/api/v1/dags/{dag_id}/tasks`	✅
获取任务细节	`/api/v1/dags/{dag_id}/tasks/{task_id}`	✅
获取任务实例	`/api/v1/dags/{dag_id}/dagRuns/{dag_run_id}/taskInstances/{task_id}`	✅
列出任务实例	`/api/v1/dags/{dag_id}/dagRuns/{dag_run_id}/taskInstances`	✅
更新任务实例	`/api/v1/dags/{dag_id}/dagRuns/{dag_run_id}/taskInstances/{task_id}`	✅
获取任务实例日志	`/api/v1/dags/{dag_id}/dagRuns/{dag_run_id}/taskInstances/{task_id}/logs/{task_try_number}`	✅
清除任务实例	`/api/v1/dags/{dag_id}/clearTaskInstances`	✅
设置任务实例状态	`/api/v1/dags/{dag_id}/updateTaskInstancesState`	✅
列出任务实例尝试次数	`/api/v1/dags/{dag_id}/dagRuns/{dag_run_id}/taskInstances/{task_id}/tries`	✅
变量
列出变量	`/api/v1/variables`	✅
创建变量	`/api/v1/variables`	✅
获取变量	`/api/v1/variables/{variable_key}`	✅
更新变量	`/api/v1/variables/{variable_key}`	✅
删除变量	`/api/v1/variables/{variable_key}`	✅
连接
列出连接	`/api/v1/connections`	✅
创建连接	`/api/v1/connections`	✅
获取连接	`/api/v1/connections/{connection_id}`	✅
更新连接	`/api/v1/connections/{connection_id}`	✅
删除连接	`/api/v1/connections/{connection_id}`	✅
测试连接	`/api/v1/connections/test`	✅
池
列出池	`/api/v1/pools`	✅
创建池	`/api/v1/pools`	✅
获取池	`/api/v1/pools/{pool_name}`	✅
更新池	`/api/v1/pools/{pool_name}`	✅
删除池	`/api/v1/pools/{pool_name}`	✅
XComs
列出 XComs	`/api/v1/dags/{dag_id}/dagRuns/{dag_run_id}/taskInstances/{task_id}/xcomEntries`	✅
获取 XCom 条目	`/api/v1/dags/{dag_id}/dagRuns/{dag_run_id}/taskInstances/{task_id}/xcomEntries/{xcom_key}`	✅
数据集
列出数据集	`/api/v1/datasets`	✅
获取数据集	`/api/v1/datasets/{uri}`	✅
获取数据集事件	`/api/v1/datasetEvents`	✅
创建数据集事件	`/api/v1/datasetEvents`	✅
获取 DAG 数据集排队事件	`/api/v1/dags/{dag_id}/dagRuns/queued/datasetEvents/{uri}`	✅
获取 DAG 数据集排队事件	`/api/v1/dags/{dag_id}/dagRuns/queued/datasetEvents`	✅
删除 DAG 数据集排队事件	`/api/v1/dags/{dag_id}/dagRuns/queued/datasetEvents/{uri}`	✅
删除 DAG 数据集排队事件	`/api/v1/dags/{dag_id}/dagRuns/queued/datasetEvents`	✅
获取数据集排队事件	`/api/v1/datasets/{uri}/dagRuns/queued/datasetEvents`	✅
删除数据集排队事件	`/api/v1/datasets/{uri}/dagRuns/queued/datasetEvents`	✅
监控
获取健康状况	`/api/v1/health`	✅
DAG 统计
获取 DAG 统计信息	`/api/v1/dags/statistics`	✅
配置
获取配置	`/api/v1/config`	✅
插件
获取插件	`/api/v1/plugins`	✅
提供商
列出提供商	`/api/v1/providers`	✅
事件日志
列出事件日志	`/api/v1/eventLogs`	✅
获取事件日志	`/api/v1/eventLogs/{event_log_id}`	✅
系统
获取导入错误	`/api/v1/importErrors`	✅
获取导入错误详情	`/api/v1/importErrors/{import_error_id}`	✅
获取健康状态	`/api/v1/health`	✅
获取版本	`/api/v1/version`	✅

配置

依赖项

此项目依赖于官方的 Apache Airflow 客户端库（apache-airflow-client）。当您安装此包时，它会自动安装。

环境变量

设置以下环境变量：

AIRFLOW_HOST=<您的 Airflow 主机>        # 可选，默认为 http://localhost:8080
AIRFLOW_API_VERSION=v1                  # 可选，默认为 v1
READ_ONLY=true                          # 可选，启用只读模式（true/false，默认为 false）

认证

选择以下认证方法之一：

基本认证（默认）：

AIRFLOW_USERNAME=<您的 Airflow 用户名>
AIRFLOW_PASSWORD=<您的 Airflow 密码>

JWT 令牌认证：

AIRFLOW_JWT_TOKEN=<您的 JWT 令牌>

要获取 JWT 令牌，您可以使用 Airflow 的认证端点：

ENDPOINT_URL="http://localhost:8080"  # 替换为您自己的 Airflow 端点
curl -X 'POST' \
  "${ENDPOINT_URL}/auth/token" \
  -H 'Content-Type: application/json' \
  -d '{ "username": "<您的用户名>", "password": "<您的密码>" }'

注意：如果同时提供了 JWT 令牌和基本认证凭据，JWT 令牌优先。

使用 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": "您的密码"
      }
    }
  }
}

JWT 令牌认证：

{
  "mcpServers": {
    "mcp-server-apache-airflow": {
      "command": "uvx",
      "args": ["mcp-server-apache-airflow"],
      "env": {
        "AIRFLOW_HOST": "https://您的 Airflow 主机",
        "AIRFLOW_JWT_TOKEN": "您的 JWT 令牌"
      }
    }
  }
}

对于只读模式（推荐安全）：

基本认证：

{
  "mcpServers": {
    "mcp-server-apache-airflow": {
      "command": "uvx",
      "args": ["mcp-server-apache-airflow"],
      "env": {
        "AIRFLOW_HOST": "https://您的 Airflow 主机",
        "AIRFLOW_USERNAME": "您的用户名",
        "AIRFLOW_PASSWORD": "您的密码",
        "READ_ONLY": "true"
      }
    }
  }
}

JWT 令牌认证：

{
  "mcpServers": {
    "mcp-server-apache-airflow": {
      "command": "uvx",
      "args": ["mcp-server-apache-airflow", "--read-only"],
      "env": {
        "AIRFLOW_HOST": "https://您的 Airflow 主机",
        "AIRFLOW_JWT_TOKEN": "您的 JWT 令牌"
      }
    }
  }
}

使用 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": "您的密码"
      }
    }
  }
}

JWT 令牌认证：

{
  "mcpServers": {
    "mcp-server-apache-airflow": {
      "command": "uv",
      "args": [
        "--directory",
        "/路径到/mcp-server-apache-airflow",
        "run",
        "mcp-server-apache-airflow"
      ],
      "env": {
        "AIRFLOW_HOST": "https://您的 Airflow 主机",
        "AIRFLOW_JWT_TOKEN": "您的 JWT 令牌"
      }
    }
  }
}

将 /路径到/mcp-server-apache-airflow 替换为您实际克隆仓库的位置。

选择 API 组

您可以通过设置 --apis 标志来选择要使用的 API 组。

uv run mcp-server-apache-airflow --apis dag --apis dagrun

默认情况下，使用所有 API。

允许的值有：

config
connections
dag
dagrun
dagstats
dataset
eventlog
importerror
monitoring
plugin
pool
provider
taskinstance
variable
xcom

只读模式

您可以通过使用 --read-only 标志或设置 READ_ONLY=true 环境变量来运行服务器的只读模式。这将仅暴露执行读操作（GET 请求）的工具，并排除任何创建、更新或删除资源的工具。

使用命令行标志：

uv run mcp-server-apache-airflow --read-only

使用环境变量：

READ_ONLY=true uv run mcp-server-apache-airflow

在只读模式下，服务器将仅暴露如下工具：

列出 DAGs、DAG 运行、任务、变量、连接等。
获取特定资源的详细信息。
读取配置和监控信息。
测试连接（非破坏性）。

写操作如创建、更新、删除 DAGs、变量、连接、触发 DAG 运行等，在只读模式下不可用。

您可以结合只读模式和 API 组选择：

uv run mcp-server-apache-airflow --read-only --apis dag --apis variable

手动执行

您也可以手动运行服务器：

make run

make run 接受以下选项：

选项：

--port: 监听 SSE 的端口（默认：8000）
--transport: 传输类型（stdio/sse/http，默认：stdio）

或者，您可以直接运行 SSE 服务器，它接受相同的参数：

make run-sse

同样，您可以使用 uv 直接启动服务，如下所示：

uv run src --transport http --port 8080

通过 Smithery 安装

要通过 Smithery 自动为 Claude Desktop 安装 Apache Airflow MCP 服务器：

npx -y @smithery/cli install @yangkyeongmo/mcp-server-apache-airflow --client claude

开发

设置开发环境

克隆仓库：

git clone https://github.com/yangkyeongmo/mcp-server-apache-airflow.git
cd mcp-server-apache-airflow

安装开发依赖项：

uv sync --dev

创建一个 .env 文件用于环境变量（开发中可选）：

touch .env

注意：运行测试不需要任何环境变量。AIRFLOW_HOST 默认为 http://localhost:8080，适用于开发和测试目的。

运行测试

该项目使用 pytest 进行测试，可用命令如下：

# 运行所有测试
make test

代码质量

# 运行代码检查
make lint

# 运行代码格式化
make format

持续集成

该项目包含一个 GitHub Actions 工作流（`.