Airflow MCP 服务器

Apache气流MCP服务器

独立社区项目；与Apache软件基金会无关或不受其认可。

通过发布 PyPI, GHCR, 和那个 MCP官方注册. 附上可复制的扫描报告、依赖性审计、SBOM和来源 每个 不可变发布.

连接克劳德、光标、VS代码或其他 主控程序 客户端连接到Apache Airflow部署，并帮助代理诊断故障的DAG。

从PagerDuty/Datadog警报粘贴气流UI链接并询问 “为什么会失败？” --代理解析URL，查找失败的任务，在MCP响应之前提取经过过滤和限制的日志错误，并可以重新触发或清除运行。写入工具带有破坏性的操作注释，MCP客户端可以使用这些注释请求确认。

亮点

• 🔍 事件响应优先 --将Airflow UI URL直接解析到失败的任务，用上下文行按错误级别过滤日志，遵循 try_number 语义正确（包括传感器）
• 🏢 多实例 -一台服务器，用于相同的API系列dev/staining/prod目标，具有per-instance凭据和拒绝未知主机的SSRF保护
• 🔒 安全控制 --选择只读模式(AIRFLOW_MCP_READ_ONLY=true)从不注册写工具；编写注释为破坏性的工具；从实例响应和操作日志中编辑配置的凭据
• 📉 代币高效 --为LLM上下文窗口设计的日志尾随、级别过滤、字节上限和截断元数据
• 🧭 气流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.11？ 使用 --with 'apache-airflow-client==2.10.0' 相反；2.10.0是最终生成的v1客户端，目标是Airflow 2的稳定API。看 气流兼容性 在使用不同的版本之前。

2.连接您的MCP客户端

最快的路径是一个完全配置了环境变量的单个实例，不需要配置文件。

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

{
  "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"
      }
    }
  }
}

{
  "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"
      }
    }
  }
}

{
  "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"
      }
    }
  }
}

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" } } }

3.问问你的经纪人

“为什么最近一轮 etl_pipeline 失败？"

"https://airflow.example.com/dags/etl_pipeline/grid这里发生了什么，可以安全地清理吗？"

“显示失败任务的错误上下文，并告诉我最小的恢复操作。”

气流兼容性

服务器通过生成的 apache-airflow-client. 对于Airflow 3，将客户端版本与您的Airflow版本相匹配：生成 模型可以在主版本中更改，并且不能保证更新的客户端 正确反序列化旧服务器的响应。气流2.11使用 最终的v1客户端版本，2.10.0，针对Airflow2的稳定API。

当 api_version 未设置，则服务器假定API与已安装的 客户(v1 对于2.x客户端， v2 对于3.x）。集 AIRFLOW_MCP_API_VERSION （或 api_version: 在注册表中）明确地 及早发现主要版本不匹配。

一个服务器进程当前只能加载一个生成的客户端主进程。全部 因此，注册表中的实例必须使用相同的API系列；单独运行 MCP服务器处理气流2和气流3。混合版本支持需要 未来的客户端适配器更改，并且今天没有宣传可以正常工作。

气流3注意事项：

• 认证：不记名代币作为JWT传递；基本凭证通过以下方式自动交换为JWT POST /auth/token 并定期刷新(AIRFLOW_MCP_TOKEN_REFRESH_SECONDS，默认值3600——将其保持在部署的JWT到期日以下，并注意401上还没有自动重新认证）。
• execution_date 订购地图 logical_date，数据集映射到资源，UI链接使用Airflow 3路线方案。工具名称和核心工作流程保持稳定；API系列的文档字段和选项可能有所不同。
• 清除气流3中不再存在的选项(include_subdags/include_parentdag，以及 include_*/reset_dag_runs 选项 airflow_clear_dag_run)被拒绝 INVALID_INPUT 而不是默默地缩小破坏性的行动。

这两个客户家族都会根据相关的拉取请求和主推送进行锻炼。我们非常欢迎来自真实Airflow部署的错误报告！

配置

单实例（仅限于环境变量）

*提供用户名+密码或令牌。

多个实例（注册表YAML）

点 AIRFLOW_MCP_INSTANCES_FILE 在YAML注册表中（它优先于单实例env变量）。值可能引用环境变量 ${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)或a ui_url --一个完整的http（s）Airflow UI URL，其主机已根据注册表解析，未知主机被拒绝（SSRF保护）。 ui_url 也自动填充 dag_id/dag_run_id/task_id 当链接包含它们时。如果两者都有 instance 和 ui_url 通过但不同意，呼叫失败 INSTANCE_MISMATCH 而不是猜测。

库贝内特斯提示： 从Secret装载注册表 /config/instances.yaml 并设置 AIRFLOW_MCP_INSTANCES_FILE=/config/instances.yaml.

服务器选项

只读模式

快速启动设置 AIRFLOW_MCP_READ_ONLY=true：写入工具（触发器、清除、， 暂停/取消暂停）从未注册。这可以防止MCP突变，但请阅读工具 仍然可以披露敏感日志、配置、渲染字段和DAG运行 数据；使用最低权限的Airflow凭据。

要有意启用恢复操作，请设置 AIRFLOW_MCP_READ_ONLY=false. 编写工具，然后携带MCP destructiveHint 客户端可以使用的注释 在决定是否请求确认时。注释是建议性的，也是 除非Airflow凭据和MCP客户端批准，否则无法启用写入 行为适合目标环境。

工具

发现和URL实用程序

阅读

写 （注释为破坏性，因此客户端可以要求批准；完全隐藏在只读模式下）

每个成功载荷都包括 request_id 用于对数相关性；失败引发结构化 ToolError 与 {code, message, request_id, context}.

事件工作流程

这就是工具设计的流程——从警报链接到四个调用中的诊断：

# 1. Alert contains an Airflow UI link → resolve it
airflow_resolve_url("https://airflow.example.com/dags/etl_pipeline/grid?dag_run_id=...")
#    → {instance, dag_id, dag_run_id, ...}

# 2. Which tasks failed in this run?
airflow_list_task_instances(dag_id="etl_pipeline", dag_run_id="scheduled__2026-01-01",
                            state=["failed"])

# 3. Get attempt metadata (authoritative try_number, retries, timings)
ti = airflow_get_task_instance(dag_id="etl_pipeline",
                               dag_run_id="scheduled__2026-01-01",
                               task_id="transform_data")

# 4. Pull only the error lines, with context, capped for the 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，以及字节/行统计数据，以便代理确切地知道它正在查看什么。主机分段日志用 --- [worker-1] --- 标题；气流3结构化日志呈现为平面线。

注意 try_number: 重新安排模式传感器可以在每次 通过Airflow 2.9重新安排；气流2.10+不再适用。始终阅读它 来自 airflow_get_task_instance 而不是猜测——推导 retries_consumed/retries_remaining 字段是启发式的。

部署

码头工人

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 .

发布映像包含lockfile的Airflow 3.3客户端，并提供 开启可流式传输的HTTP :8765 (/mcp 端点， /health 用于探头）。MCP HTTP 端点没有内置的调用者身份验证：保持其环回绑定或放置 它位于经过身份验证的私有代理之后。安装相同的API家族注册表YAML 用于多实例设置。气流2部署应使用固定的本地 安装路径如上，直到发布单独的v1映像。

CI审核确切映像的已安装Python依赖关系，并扫描 带有固定Cisco MCP扫描仪的只读和可写MCP工具表面 发布的YARA分析仪。安全工作流在扫描不完整或任何情况下失败 未经诊断的YARA发现。从v1.0.1开始，发布资产包括 机器可读的扫描报告和发布图像清单附有SBOM 以及涵盖其更广泛包装库存的来源证明。这些 是自动检查，而不是安全认证或替代 部署特定审查。

FastMCP工具

A. fastmcp.json 包含在内，因此FastMCP感知工具可以自动发现入口点和部署默认值。

发展

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针对两者运行单元套件 apache-airflow-client Python 3.10–3.13上的家庭。相关的拉取请求、主推送、夜间运行和发布也执行了实时停靠的气流2.11和3.3。看 贡献.md 指南和 代理商.md 如果你把编码代理指向这个仓库（它是为此而写的）。

贡献

欢迎提出问题和PR，特别是：

• 来自真实气流事件响应工作流程和版本组合的报告
• 有限日志、诊断安全和URL优先工作流改进
• 更多MCP主机和部署类型的客户端设置配方

如果此服务器为您保存了调试会话⭐ 帮助其他Airflow团队找到它。

许可证

Apache 2.0——请参阅 许可证.

Apache Airflow和Airflow是Apache软件基金会的注册商标。该独立项目不隶属于ASF或得到ASF的认可。