Apache Airflow MCP Server

气流MCP服务器

      

运行和使用Apache Airflow MCP服务器的人类入口点。此服务器公开了安全、专注的工具来检查Airflow DAG、运行和日志（可选的写入操作由客户端批准控制）。响应是结构化的JSON对象（字典），包括 request_id 为了可追溯性。

快速入门

1） 安装服务器（PyPI）

安装时使用 uv 因此，您正在练习交付给用户的确切部分，并获得可重复的虚拟环境：

uv tool install apache-airflow-mcp-server

不 uv？回落到pip：

pip install apache-airflow-mcp-server

2） 配置实例（必需）

集 AIRFLOW_MCP_INSTANCES_FILE 到一个列出可用Airflow实例的YAML文件。值可以使用以下方式引用环境变量 ${VAR} 语法。缺少变量会导致启动错误。

示例(examples/instances.yaml):

# Data team staging instance
data-stg:
  host: https://airflow.data-stg.example.com/
  api_version: v1
  verify_ssl: true
  auth:
    type: basic
    username: ${AIRFLOW_INSTANCE_DATA_STG_USERNAME}
    password: ${AIRFLOW_INSTANCE_DATA_STG_PASSWORD}

# ML team staging instance
ml-stg:
  host: https://airflow.ml-stg.example.com/
  api_version: v1
  verify_ssl: true
  auth:
    type: basic
    username: ${AIRFLOW_INSTANCE_ML_STG_USERNAME}
    password: ${AIRFLOW_INSTANCE_ML_STG_PASSWORD}

# Bearer token (experimental)
# ml-prod:
#   host: https://airflow.ml-prod.example.com/
#   api_version: v1
#   verify_ssl: true
#   auth:
#     type: bearer
#     token: ${AIRFLOW_INSTANCE_ML_PROD_TOKEN}

承载令牌身份验证是 实验性的基本身份验证仍然是经过充分测试的主要途径。

Kubernetes部署提示： 提供 instances.yaml 通过一个秘密并将其安装在 /config/instances.yaml （套 AIRFLOW_MCP_INSTANCES_FILE=/config/instances.yaml).

环境变量：

• AIRFLOW_MCP_INSTANCES_FILE （必填）：注册表YAML的路径
• AIRFLOW_MCP_DEFAULT_INSTANCE （可选）：默认实例密钥
• AIRFLOW_MCP_HTTP_HOST （默认值：127.0.0.1）
• AIRFLOW_MCP_HTTP_PORT （默认值：8765）
• AIRFLOW_MCP_TIMEOUT_SECONDS （默认值：30）
• AIRFLOW_MCP_LOG_FILE （可选）
• AIRFLOW_MCP_HTTP_BLOCK_GET_ON_MCP （默认值：true）

3） 运行服务器

• HTTP（推荐用于工具）：

uv run airflow-mcp --transport http --host 127.0.0.1 --port 8765

• STDIO（命令行界面/终端工作流）：

uv run airflow-mcp --transport stdio

健康检查（HTTP）： GET /health → 200 OK.

提示：A fastmcp.json 包含用于FastMCP工具的发现/配置：

{
  "$schema": "https://gofastmcp.com/schemas/fastmcp_config/v1.json",
  "entrypoint": { "file": "src/airflow_mcp/server.py", "object": "mcp" },
  "deployment": { "transport": "http", "host": "127.0.0.1", "port": 8765 }
}

4） 典型事件工作流程

从Airflow UI URL开始（通常在Datadog警报中）：

1. airflow_resolve_url(url) → 解决 instance, dag_id, dag_run_id, task_id.
2. airflow_list_dag_runs(instance|ui_url, dag_id) → 确认最近的状态。
3. airflow_get_task_instance(instance|ui_url, dag_id, dag_run_id, task_id, include_rendered?, max_rendered_bytes?) → 检查任务元数据、尝试次数和可选的呈现字段。
4. airflow_get_task_instance_logs(instance|ui_url, dag_id, dag_run_id, task_id, try_number, filter_level?, context_lines?, tail_lines?, max_bytes?) → 通过可选的过滤和截断来检查故障。

所有工具均接受 instance 或 ui_url。如果两者都给出但不同意，则通话失败 INSTANCE_MISMATCH. ui_url 必须是完全限定的http气流URL；使用 airflow_list_instances() 在只有实例密钥时发现有效主机。

工具参考（结构化JSON）

发现和URL实用程序：

• airflow_list_instances() → 列出已配置的实例密钥和默认值
• airflow_describe_instance(instance) → 主机、api_version、verify_sl， auth_type （已编辑）
• airflow_resolve_url(url) → 从Airflow UI URL解析实例和标识符

只读工具：

• airflow_list_dags(instance|ui_url, limit?, offset?, state?/filters) → 带有UI链接的紧凑型DAGs
• airflow_get_dag(instance|ui_url, dag_id) → DAG详细信息+UI链接
• airflow_list_dag_runs(instance|ui_url, dag_id, state?, limit?, offset?, order_by?, descending?) → runs+每次运行的UI链接。默认为 execution_date 降序（“最新优先”）。通过以下方式接受显式排序 start_date, end_date，或 execution_date
• airflow_get_dag_run(instance|ui_url, dag_id, dag_run_id) → 运行详细信息+UI链接
• airflow_list_task_instances(instance|ui_url, dag_id, dag_run_id, limit?, offset?, state?, task_ids?) → 运行的任务尝试，包括每次尝试的日志URL、可选的按状态或任务id的服务器端筛选，以及 filters 回声描述应用滤波器(count 反映过滤后的结果； total_entries 镜像API响应（如果可用）
• airflow_get_task_instance(instance|ui_url, dag_id, dag_run_id, task_id, include_rendered?, max_rendered_bytes?) → 简洁的任务元数据（状态、时间、尝试、配置），带有可选的渲染模板字段和直接UI链接；传感器增量 try_number 在每次重新安排时，请将其视为尝试索引（派生 retries_* 字段是启发式的）
• airflow_get_task_instance_logs(instance|ui_url, dag_id, dag_run_id, task_id, try_number, filter_level?, context_lines?, tail_lines?, max_bytes?) → 日志文本，可选过滤；响应包括 truncated, auto_tailed，以及统计数据
• airflow_dataset_events(instance|ui_url, dataset_uri, limit?) → 数据集事件（可选功能）

编写工具（需要客户批准；破坏性）：

• airflow_trigger_dag(instance|ui_url, dag_id, conf?, logical_date?, dag_run_id?, note?)
• airflow_clear_task_instances(instance|ui_url, dag_id, task_ids?, start_date?, end_date?, include_*?, dry_run?)
• airflow_clear_dag_run(instance|ui_url, dag_id, dag_run_id, include_*?, dry_run?, reset_dag_runs?)
• airflow_pause_dag(instance|ui_url, dag_id) / airflow_unpause_dag(instance|ui_url, dag_id)

合同：

• 成功：dict有效载荷包括 request_id （可在日志中追踪）。FastMCP会自动对它们进行序列化。
• 故障：工具引发MCP ToolError 其有效载荷仍然是一个紧凑的JSON字符串

{ "code": "INVALID_INPUT", "message": "...", "request_id": "...", "context"?: {...} }.

日志筛选(airflow_get_task_instance_logs)

通过服务器端过滤和规范化有效载荷进行高效的事件分类：

典型事件工作流程：

# Example: Find errors in recent execution with context
airflow_get_task_instance_logs(
    dag_id="etl_pipeline",
    dag_run_id="scheduled__2025-10-30",
    task_id="transform_data",
    try_number=2,
    tail_lines=500,       # Last 500 lines only
    filter_level="error", # Show ERROR, CRITICAL, FATAL, Exception, Traceback
    context_lines=5       # Include 5 lines before/after each error
)

两种呼叫模式（需要 try_number):

1. ti = airflow_get_task_instance(...) → read ti["attempts"]["try_number"].
2. airflow_get_task_instance_logs(..., try_number=ti["attempts"]["try_number"]).

传感器和重新安排可以增加 try_number 而不消耗重试时间；保持这种明确性可以防止服务器对“最新尝试”做出错误的假设

传感器将每次重新安排视为另一次尝试，因此 try_number 最好解释为 尝试指数 而不是“重试次数” retries_consumed / retries_remaining 字段是基于配置的重试的启发式方法，可能不符合Airflow对长时间运行的传感器的概念——如果需要权威计数，请始终检查任务元数据。

参数：

• filter_level: "error" （严格）| "warning" （包括错误）| "info" （所有级别）
• context_lines：对称上下文（每次匹配前的N+匹配后的N），限制为\[0，1000\]
• tail_lines：在滤波之前提取最后N行，夹紧到\[0，100K\]
• max_bytes：硬上限（默认值：100KB≈25K令牌）
• 以浮点数/字符串形式提供的数字输入在服务器端被强制和夹紧，因此客户端可以安全地传递用户输入的值。

响应形状：

• log：单个规范化字符串。当Airflow API返回主机分段日志时，表单的头 --- [worker-1] --- （或 --- [unknown-host] ---)在段之间插入空行，以便LLM可以推断执行位置。
• truncated: true 如果输出超过 max_bytes
• auto_tailed: true 如果日志>100MB，则触发自动尾到最后10K行
• match_count：匹配的行数 filter_level （在上下文扩展之前）
• meta.filters：应用有效滤波器的回声（显示箝位值）
• bytes_returned, original_lines, returned_lines：与客户端看到的规范化文本对齐的统计数据

任务发现失败(airflow_list_task_instances)

airflow_list_task_instances 现在公开了服务器端过滤器，使 airflow_get_failed_task_instance 助手是不必要的。对每个对象使用相同的图元 失败的任务工作流：

# All failed tasks for a known run (count reflects filtered results)
failed = airflow_list_task_instances(
    dag_id="etl_pipeline",
    dag_run_id="scheduled__2025-10-30",
    state=["failed"]
)

# Failed tasks for a subset of task IDs (case: only sensors + downloaders)
subset = airflow_list_task_instances(
    dag_id="etl_pipeline",
    dag_run_id="backfill__2025-10-30",
    state=["failed"],
    task_ids=["check_source", "download_payload"]
)

# Recipe: latest failed run → failed tasks
latest_failed = airflow_list_dag_runs(
    dag_id="etl_pipeline",
    state=["failed"],
    limit=1
)["dag_runs"][0]
failed_tasks = airflow_list_task_instances(
    dag_id="etl_pipeline",
    dag_run_id=latest_failed["dag_run_id"],
    state=["failed"]
)

为什么要改变？

• 单个工具保持可组合性：按以下条件过滤 state, task_ids或两者皆有。
• 回应包括 filters （回声）和 total_entries 当API共享它时，调用方可以检测是否需要额外的分页，同时仍然获得筛选 count.
• 以前依赖的代理商 airflow_get_failed_task_instance 应该迁移到 airflow_list_task_instances(state=["failed"]) （可选择在前面加上 airflow_list_dag_runs 以解决相关问题）。

任务实例元数据(airflow_get_task_instance)

将元数据与日志配对，以实现更快的分类：

# Example: Inspect failed task metadata and rendered fields
task_meta = airflow_get_task_instance(
    dag_id="etl_pipeline",
    dag_run_id="scheduled__2025-10-30",
    task_id="transform_data",
    include_rendered=True,
    max_rendered_bytes=100_000,
)

# Use try_number + ui_url.log to pull logs next
logs = airflow_get_task_instance_logs(
    dag_id="etl_pipeline",
    dag_run_id="scheduled__2025-10-30",
    task_id=task_meta["task_instance"]["task_id"],
    try_number=task_meta["attempts"]["try_number"],
    filter_level="error",
)

亮点：

• task_instance：状态、主机、操作员、计时（计算 duration_ms)
• task_config：所有者、重试次数、重试延迟（如果可用）
• attempts：当前尝试次数，已消耗/剩余的重试次数
• rendered_fields （可选）：带有字节上限的渲染模板值(max_rendered_bytes，默认100KB；截断的有效载荷返回 { "_truncated": "Increase max_rendered_bytes" })
• ui_url：直接 grid 和 log 任务尝试的链接

客户备注

• MCP客户端（例如IDE）应使用JSON参数按名称调用工具并处理JSON响应。
• 对于基于URL的调用，服务器会根据配置的注册表（SSRF保护）验证主机名。
• 写入工具带有注释，因此MCP客户端将在执行前提示确认。

添加到MCP客户端

游标–stdio（按需生成）

向添加条目 ~/.cursor/mcp.json.Cursor将生成服务器进程并通过stdio进行通信。

{
  "mcpServers": {
    "airflow-stdio": {
      "command": "uv",
      "args": [
        "--directory",
        "/path/to/apache-airflow-mcp-server",
        "run",
        "airflow-mcp",
        "--transport",
        "stdio"
      ],
      "env": {
        "AIRFLOW_MCP_INSTANCES_FILE": "/path/to/instances.yaml",
        "AIRFLOW_MCP_DEFAULT_INSTANCE": "production"
      }
    }
  }
}

笔记：

• 替换 /path/to/apache-airflow-mcp-server 和 /path/to/instances.yaml 与你的道路。
• 不需要长时间运行的服务器；游标在每个会话中启动进程。

游标–本地HTTP

自己运行服务器，然后将Cursor指向HTTP端点。

1. 启动服务器：

uv run airflow-mcp --transport http --host 127.0.0.1 --port 8765

1. 在 ~/.cursor/mcp.json:

{
  "mcpServers": {
    "airflow-local-http": {
      "url": "http://127.0.0.1:8765/mcp"
    }
  }
}

游标–远程HTTP

将光标指向已部署的服务器。如果前面有身份验证代理，则可以添加可选标头。

{
  "mcpServers": {
    "airflow-remote-http": {
      "url": "https://airflow-mcp.internal.example.com/mcp",
      "headers": {
        "X-API-Key": "your-token-if-proxied"
      }
    }
  }
}

其他客户

• 许多MCP客户端接受stdio命令或HTTP URL；上述示例具有普遍性。
• A. fastmcp.json 包含用于可自动发现入口点的FastMCP感知工具。

发展

# Install dependencies
uv sync

# Run tests
uv run pytest

# Run linter
uv run ruff check .

# Run the server (stdio)
uv run airflow-mcp --transport stdio

# Run the server (HTTP)
uv run airflow-mcp --transport http --host 127.0.0.1 --port 8765

测试策略：

• 没有真正的网络呼叫；测试补丁 airflow_mcp.client_factory._import_airflow_client.
• 重置设备中的注册表缓存；断言 request_id 存在、URL优先级和结构化日志字段。

可观察性：

• 结构化日志 tool_start, tool_success/tool_error, duration_ms, response_bytes，以及上下文字段。

许可证

Apache 2.0-请参阅 许可证 了解详情。