madamak/apache-airflow-mcp-server

# Apache Airflow MCP Server [](https://modelcontextprotocol.io) [](https://pypi.org/project/apache-airflow-mcp-server/) [](https://pypi.org/project/apache-airflow-mcp-server/) [](https://airflow.apache.org/) [](https://github.com/madamak/apache-airflow-mcp-server/actions/workflows/ci.yml) [](https://github.com/madamak/apache-airflow-mcp-server/actions/workflows/security.yml) [](https://opensource.org/licenses/Apache-2.0) [](https://github.com/astral-sh/ruff) 通过 [PyPI](https://pypi.org/project/apache-airflow-mcp-server/)， [GHCR](https://github.com/madamak/apache-airflow-mcp-server/pkgs/container/apache-airflow-mcp-server)， 以及 [官方 MCP Registry](https://registry.modelcontextprotocol.io/v0.1/servers?search=io.github.madamak%2Fapache-airflow-mcp-server) 发布。 每个[不可变发布](https://github.com/madamak/apache-airflow-mcp-server/releases/latest)中都附带有可重现的扫描报告、依赖审计、SBOMs 和来源证明。 将 Claude、Cursor、VS Code 或其他 [MCP](https://modelcontextprotocol.io) 客户端连接到您的 Apache Airflow 部署，帮助 agent 诊断失败的 DAG。 粘贴来自 PagerDuty/Datadog 告警的 Airflow UI 链接，并询问*“为什么这个失败了？”* — agent 会解析 URL，找到失败的任务，在返回 MCP 响应之前拉取经过过滤和限制大小的日志错误，并且可以重新触发或清除运行。写入工具带有破坏性操作标注，MCP 客户端可利用该标注来请求确认。 ## 核心亮点 - 🔍 **优先考虑事件响应** — 直接将 Airflow UI URL 解析到失败任务，按带有上下文行的错误级别过滤日志，正确遵循 `try_number` 语义（包括 sensors） - 🏢 **多实例支持** — 一个服务器可用于相同 API 系列的开发/测试/生产目标，支持每个实例使用单独的凭据，并带有拒绝未知主机的 SSRF 防护 - 🔒 **安全控制** — 可选的只读模式（`AIRFLOW_MCP_READ_ONLY=true`）从不注册写入工具；写入工具标注为破坏性操作；配置的凭据会从实例响应和操作日志中脱敏处理 - 📉 **Token 高效** — 日志追踪、级别过滤、字节上限和截断元数据专为 LLM 上下文窗口设计 - 🧭 **支持 Airflow 2 和 3** — 包含 API v1/v2 适配器，针对 Airflow 2.11 和 3.3 进行了实时的 E2E 测试，包括 JWT 认证 - 📎 **可追踪** — 每个响应都带有一个与结构化服务器日志匹配的 `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 ``` ### 2. 连接您的 MCP 客户端 最快的方法是完全通过环境变量配置单个实例 — 无需配置文件。

Claude Code

``` 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 ```

Claude Desktop

添加到 `claude_desktop_config.json` (Settings → Developer → Edit Config)： ``` { "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" } } } } ```

Cursor

添加到 `~/.cursor/mcp.json`： ``` { "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" } } } } ```

VS Code (Copilot)

添加到 `.vscode/mcp.json`： ``` { "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" } } } } ```

任何客户端，通过 HTTP

自行运行服务器并让客户端指向该 endpoint： ``` 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" } } } ``` 健康检查：`GET /health` → `200 OK`。

### 3. 向您的 agent 提问 ## Airflow 兼容性 服务器通过生成的 `apache-airflow-client` 与 Airflow 通信。 对于 Airflow 3，请将客户端版本与您的 Airflow 版本匹配：生成的 模型可能会在同一个大版本内发生变化，并且不保证较新的客户端能 正确反序列化较旧服务器的响应。Airflow 2.11 使用 最终的 v1 客户端版本 2.10.0，以对接 Airflow 2 的稳定 API。 | 您的 Airflow | REST API | 安装命令 | 实时 E2E 状态 | |---|---|---|---| | 3.3 | v2 | `uv tool install apache-airflow-mcp-server --with 'apache-airflow-client==3.3.0'` | ✅ 3.3.0 | | 2.11 | v1 | `uv tool install apache-airflow-mcp-server --with 'apache-airflow-client==2.10.0'` | ✅ Airflow 2.11 + 最终的 v1 客户端 2.10.0 | | 3.0–3.2 | v2 | 将客户端固定为您部署的 Airflow 3 版本 | 🧪 不在当前的实时测试矩阵中 | | 2.5–2.10 | v1 | 使用最终的 v1 客户端，`apache-airflow-client==2.10.0` | 🧪 不在当前的实时测试矩阵中 | 当未设置 `api_version` 时，服务器会假定 API 与已安装的 客户端匹配（2.x 客户端对应 `v1`，3.x 客户端对应 `v2`）。显式设置 `AIRFLOW_MCP_API_VERSION`（或在注册表中的 `api_version:`） 以便尽早捕获大版本不匹配的情况。 目前一个服务器进程只能加载一个生成的客户端主版本。因此，注册表中的所有 实例必须使用相同的 API 系列；请为 Airflow 2 和 Airflow 3 分别运行独立的 MCP 服务器进程。混合版本支持需要 未来的客户端适配器更改，目前并不保证可用。 Airflow 3 说明： - **认证**：bearer token 作为 JWT 传递；basic 凭据会通过 `POST /auth/token` 自动换取 JWT，并定期刷新（`AIRFLOW_MCP_TOKEN_REFRESH_SECONDS`，默认为 3600 — 请保持其低于您部署的 JWT 过期时间，并注意目前尚无针对 401 的自动重新认证）。 - `execution_date` 排序映射到 `logical_date`，datasets 映射到 assets，UI 链接使用 Airflow 3 路由方案。工具名称和核心工作流保持稳定；但文档化的字段和选项可能会因 API 系列而异。 - 对于在 Airflow 3 中不再存在的清除选项（`include_subdags`/`include_parentdag`，以及 `airflow_clear_dag_run` 的 `include_*`/`reset_dag_runs` 选项），系统将返回 `INVALID_INPUT` 拒绝执行，而不是静默地缩小破坏性操作的范围。 两个客户端系列都会在相关的 pull request 和主分支推送时进行测试。非常欢迎提供来自真实 Airflow 部署的 Bug 报告！ ## 配置 ### 单实例（仅限环境变量） | 变量 | 必填 | 描述 | |---|---|---| | `AIRFLOW_MCP_HOST` | ✅ | Airflow base URL，例如 `https://airflow.example.com` | | `AIRFLOW_MCP_USERNAME` / `AIRFLOW_MCP_PASSWORD` | ✅* | Basic 认证凭据 | | `AIRFLOW_MCP_TOKEN` | ✅* | Bearer/JWT token（用于替代 basic auth） | | `AIRFLOW_MCP_API_VERSION` | | `v1` (Airflow 2) 或 `v2` (Airflow 3)；默认匹配已安装的 `apache-airflow-client` | | `AIRFLOW_MCP_VERIFY_SSL` | | 验证 TLS 证书（默认 `true`） | \* 提供用户名+密码或 token 之一。 ### 多实例（注册表 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 防护）。如果链接中包含相关参数，`ui_url` 还会自动填充 `dag_id`/`dag_run_id`/`task_id`。如果同时传递了 `instance` 和 `ui_url` 且内容不一致，调用将因 `INSTANCE_MISMATCH` 失败，而不是盲目猜测。 **Kubernetes 技巧：** 将来自 Secret 的注册表挂载到 `/config/instances.yaml`，并设置 `AIRFLOW_MCP_INSTANCES_FILE=/config/instances.yaml`。 ### 服务器选项 | 变量 | 默认值 | 描述 | |---|---|---| | `AIRFLOW_MCP_DEFAULT_INSTANCE` | | 默认实例键（同时命名环境变量实例） | | `AIRFLOW_MCP_READ_ONLY` | `false` | 完全不注册写入工具 | | `AIRFLOW_MCP_HTTP_HOST` / `AIRFLOW_MCP_HTTP_PORT` | `127.0.0.1` / `8765` | HTTP transport 绑定 | | `AIRFLOW_MCP_TIMEOUT_SECONDS` | `30` | Airflow API 超时时间 | | `AIRFLOW_MCP_TOKEN_REFRESH_SECONDS` | `3600` | Airflow 3：basic-auth 实例的 JWT 刷新间隔 | | `AIRFLOW_MCP_LOG_FILE` | | 可选的日志文件路径 | | `AIRFLOW_MCP_ENABLE_EXTENDED_CLEAR_PARAMS` | `false` | 启用 `include_*` 清除参数 (Airflow ≥2.6) | | `AIRFLOW_MCP_HTTP_BLOCK_GET_ON_MCP` | `true` | 在 HTTP 部署中对 `GET /mcp` (读取 SSE) 返回 405 | ### 只读模式 快速入门中设置了 `AIRFLOW_MCP_READ_ONLY=true`：从不注册写入工具（触发、清除、 暂停/取消暂停）。这可以防止 MCP 突变，但读取工具 仍可能泄露敏感日志、配置、渲染字段和 DAG 运行 数据；请使用最小权限的 Airflow 凭据。 要特意启用恢复操作，请设置 `AIRFLOW_MCP_READ_ONLY=false`。 写入工具随后会带有 MCP `destructiveHint` 标注，客户端在 决定是否请求确认时可以使用该标注。标注仅供参考，因此除非 Airflow 凭据和 MCP 客户端的审批 行为适合目标环境，否则不要启用写入操作。 ## 工具 **发现与 URL 工具** | 工具 | 描述 | |---|---| | `airflow_list_instances` | 列出已配置的实例键及默认实例 | | `airflow_describe_instance` | 主机、API 版本、认证类型（机密信息已脱敏） | | `airflow_resolve_url` | 将 Airflow UI URL 解析为实例 + dag/run/task 标识符 | **读取** | 工具 | 描述 | |---|---| | `airflow_list_dags` | 带有暂停状态和 UI 链接的 DAG | | `airflow_get_dag` | DAG 详情 | | `airflow_list_dag_runs` | 带有状态过滤和排序的运行（默认最新优先） | | `airflow_get_dag_run` | 单次运行的详情 | | `airflow_list_task_instances` | 某次运行的任务尝试；在服务端按 `state` / `task_ids` 过滤 | | `airflow_get_task_instance` | 任务元数据、重试次数、计时，可选的渲染模板字段 | | `airflow_get_task_instance_logs` | 带有级别过滤、追踪、上下文行和字节上限的日志 | | `airflow_dataset_events` | Dataset (Airflow 2) / asset (Airflow 3) 事件 | **写入**（标注为破坏性，客户端可要求批准；在只读模式下完全隐藏） | 工具 | 描述 | |---|---| | `airflow_trigger_dag` | 触发带有可选 conf/logical date/note 的运行 | | `airflow_clear_task_instances` | 跨运行清除任务实例（默认 `dry_run=true`） | | `airflow_clear_dag_run` | 清除整个运行（默认 `dry_run=true`） | | `airflow_pause_dag` / `airflow_unpause_dag` | 切换 DAG 调度 | 每个成功负载都包含用于日志关联的 `request_id`；失败会引发结构化的 `ToolError`，包含 `{code, message, request_id, context}`。 ## 事件工作流 工具设计所围绕的流程 — 在四次调用内从告警链接获得诊断： ``` # 1. 告警包含 Airflow UI 链接 → 解析它 airflow_resolve_url("https://airflow.example.com/dags/etl_pipeline/grid?dag_run_id=...") # → {instance, dag_id, dag_run_id, ...} # 2. 此次运行中有哪些任务失败了？ airflow_list_task_instances(dag_id="etl_pipeline", dag_run_id="scheduled__2026-01-01", state=["failed"]) # 3. 获取尝试元数据（权威 try_number、重试次数、时间） ti = airflow_get_task_instance(dag_id="etl_pipeline", dag_run_id="scheduled__2026-01-01", task_id="transform_data") # 4. 仅提取错误行及其上下文，并为 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`（>100MB 的日志自动追踪尾部）、`match_count` 以及字节/行数统计，以便 agent 准确知道它在查看什么。基于主机分段的日志会通过 `--- [worker-1] ---` 头部进行扁平化；Airflow 3 结构化日志呈现为纯文本行。 ## 部署 ### 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` endpoint，`/health` 用于探测）。MCP HTTP endpoint 没有内置的调用方身份验证：请将其保持为环回绑定，或将其 置于经过身份验证的私有代理之后。为多实例设置挂载同属一个 API 系列的注册表 YAML。 Airflow 2 部署应使用上述固定的本地 安装路径，直到发布单独的 v1 镜像。 CI 会审计该镜像安装的确切 Python 依赖项，并使用固定的 Cisco MCP Scanner 版本的 YARA 分析器扫描只读和启用写入的 MCP 工具接口。如果扫描不完整或发现任何 未处理的 YARA 结果，安全工作流就会失败。从 v1.0.1 开始，发布资产包含 机器可读的扫描报告，并且发布镜像清单带有附带的 SBOM 及其更广泛包清单的来源证明。这些 是自动化检查，并非安全认证，也不能替代针对特定部署的审查。 ### FastMCP 工具 包含了一个 `fastmcp.json`，以便支持 FastMCP 的工具能自动发现 entrypoint 和部署默认设置。 ## 开发 ``` 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` 的两个系列运行单元测试套件。相关的 pull request、主分支推送、每日定时运行和发布也会对实时的 Docker 化 Airflow 2.11 和 3.3 进行测试。有关指南，请参阅 [CONTRIBUTING.md](CONTRIBUTING.md)；如果您要将编码 agent 指向此代码库（它正是为此编写的），请参阅 [AGENTS.md](AGENTS.md)。 ## 许可证 Apache 2.0 — 详见 [LICENSE](LICENSE)。 Apache Airflow 和 Airflow 是 The Apache Software Foundation 的注册商标。此独立项目不隶属于 ASF，也未获得其认可。

标签：Apache Airflow, DAG, MCP服务器, Python, 故障诊断, 无后门, 请求拦截, 运维工具, 逆向工具