mcp-server-apache-airflow

这是一个为 Apache Airflow 实现的模型上下文协议（MCP）服务器，使得与 MCP 客户端的无缝集成成为可能。该项目提供了一种通过模型上下文协议与 Apache Airflow 进行交互的标准化方式。

关于

该项目实现了一个模型上下文协议服务器，该服务器封装了 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}/clearTaskInstances	✅
设置任务实例状态	/api/v1/dags/{dag_id}/updateTaskInstancesState	✅
变量
列出变量	/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-主机>
AIRFLOW_USERNAME=<你的-airflow-用户名>
AIRFLOW_PASSWORD=<你的-airflow-密码>
AIRFLOW_API_VERSION=v1  # 可选，默认为 v1

与 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",
        "/path/to/mcp-server-apache-airflow",
        "run",
        "mcp-server-apache-airflow"
      ],
      "env": {
        "AIRFLOW_HOST": "https://你的-airflow-主机",
        "AIRFLOW_USERNAME": "你的用户名",
        "AIRFLOW_PASSWORD": "你的密码"
      }
    }
  }
}

将 /path/to/mcp-server-apache-airflow 替换为你克隆仓库的实际路径。

选择 API 组

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

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

默认是使用所有 API。

允许的值为：

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

手动执行

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

make run

make run 接受以下选项：

选项：

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

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

make run-sse

通过 Smithery 安装

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

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

贡献

欢迎贡献！请随时提交 Pull Request。

当 pyproject.toml 中的 project.version 更新时，包会自动部署到 PyPI。请遵循语义化版本控制进行版本管理。

请在 PR 中包含版本更新，以便将更改应用到核心逻辑。

许可证

MIT 许可证