mcp-server-apache-airflow

mcp-server-apache-airflow

一个用于 Apache Airflow 的模型上下文协议（MCP）服务实现，可与 MCP 客户端无缝集成。本项目提供了一种通过模型上下文协议与 Apache Airflow 交互的标准化方式。

关于

本项目实现了一个模型上下文协议服务，用于封装 Apache Airflow 的 REST API，使 MCP 客户端能够以标准化方式与 Airflow 交互。项目使用 Apache Airflow 官方客户端库，以确保兼容性和可维护性。

功能实现状态

功能	API 路径	状态
DAG 管理
列出 DAG	/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}	✅
批量修补 DAG	/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}	✅
XCom
列出 XCom	/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=<your-airflow-host>        # Optional, defaults to http://localhost:8080
AIRFLOW_API_VERSION=v1                  # Optional, defaults to v1
READ_ONLY=true                          # Optional, enables read-only mode (true/false, defaults to false)

身份验证

选择以下一种身份验证方式：

基本身份验证（默认）：

AIRFLOW_USERNAME=<your-airflow-username>
AIRFLOW_PASSWORD=<your-airflow-password>

JWT 令牌身份验证：

AIRFLOW_JWT_TOKEN=<your-jwt-token>

若要获取 JWT 令牌，可以使用 Airflow 的身份验证端点：

ENDPOINT_URL="http://localhost:8080"  # Replace with your Airflow endpoint
curl -X 'POST' \
  "${ENDPOINT_URL}/auth/token" \
  -H 'Content-Type: application/json' \
  -d '{ "username": "<your-username>", "password": "<your-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://your-airflow-host",
        "AIRFLOW_USERNAME": "your-username",
        "AIRFLOW_PASSWORD": "your-password"
      }
    }
  }
}

JWT 令牌身份验证：

{
  "mcpServers": {
    "mcp-server-apache-airflow": {
      "command": "uvx",
      "args": ["mcp-server-apache-airflow"],
      "env": {
        "AIRFLOW_HOST": "https://your-airflow-host",
        "AIRFLOW_JWT_TOKEN": "your-jwt-token"
      }
    }
  }
}

若要启用只读模式（出于安全考虑，建议使用）：

基本身份验证：

{
  "mcpServers": {
    "mcp-server-apache-airflow": {
      "command": "uvx",
      "args": ["mcp-server-apache-airflow"],
      "env": {
        "AIRFLOW_HOST": "https://your-airflow-host",
        "AIRFLOW_USERNAME": "your-username",
        "AIRFLOW_PASSWORD": "your-password",
        "READ_ONLY": "true"
      }
    }
  }
}

JWT 令牌身份验证：

{
  "mcpServers": {
    "mcp-server-apache-airflow": {
      "command": "uvx",
      "args": ["mcp-server-apache-airflow", "--read-only"],
      "env": {
        "AIRFLOW_HOST": "https://your-airflow-host",
        "AIRFLOW_JWT_TOKEN": "your-jwt-token"
      }
    }
  }
}

使用 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://your-airflow-host",
        "AIRFLOW_USERNAME": "your-username",
        "AIRFLOW_PASSWORD": "your-password"
      }
    }
  }
}

JWT 令牌身份验证：

{
  "mcpServers": {
    "mcp-server-apache-airflow": {
      "command": "uv",
      "args": [
        "--directory",
        "/path/to/mcp-server-apache-airflow",
        "run",
        "mcp-server-apache-airflow"
      ],
      "env": {
        "AIRFLOW_HOST": "https://your-airflow-host",
        "AIRFLOW_JWT_TOKEN": "your-jwt-token"
      }
    }
  }
}

将 /path/to/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

在只读模式下，服务只会公开以下工具：

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

在只读模式下，无法使用创建、更新、删除 DAG、变量、连接以及触发 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 Server：

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

开发

设置开发环境

1. 克隆仓库：

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

2. 安装开发依赖项：

uv sync --dev

3. 创建用于设置环境变量的 .env 文件（开发时可选）：

touch .env

注意：运行测试无需设置环境变量。开发和测试时，AIRFLOW_HOST 默认值为 http://localhost:8080。

运行测试

本项目使用 pytest 进行测试，可使用以下命令：

# Run all tests
make test

代码质量

# Run linting
make lint

# Run code formatting
make format

持续集成

本项目包含一个 GitHub Actions 工作流（.github/workflows/test.yml），会自动执行以下操作：

• 在 Python 3.10、3.11 和 3.12 上运行测试
• 使用 ruff 执行代码检查
• 在每次向 main 分支推送或提交拉取请求时运行

CI 流水线会在合并任何更改之前，确保代码质量并验证其与受支持的 Python 版本兼容。

贡献

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

更新 pyproject.toml 中的 project.version 后，软件包会自动部署到 PyPI。 版本号请遵循语义化版本规范。

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

许可证

MIT 许可证