JinDaGe - mcp-server-airflow-token MCP Details

article

README

🚀 mcp-server-airflow-token

这是一个支持Bearer令牌认证的Apache Airflow的模型上下文协议（MCP）服务器，可实现与Astronomer Cloud和独立Airflow实例的无缝集成。

基于Gyeongmo Nathan Yang的mcp-server-apache-airflow

此分支通过添加Bearer令牌认证支持对原始MCP服务器进行了增强，使其与Astronomer Cloud和其他基于令牌的Airflow部署兼容。

✨ 主要特性

✅ Bearer令牌认证 - 现代Airflow部署的主要认证方法
✅ 与Astronomer Cloud兼容 - 可与Astronomer的托管Airflow无缝协作
✅ 向后兼容 - 仍然支持用户名/密码认证
✅ 增强的URL处理 - 正确处理诸如/deployment-id之类的部署路径

📚 详细文档

本项目实现了一个模型上下文协议服务器，该服务器封装了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}/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=<your-airflow-host>        # 可选，默认为 http://localhost:8080
AIRFLOW_TOKEN=<your-airflow-api-token>  # 你的Airflow API令牌
AIRFLOW_API_VERSION=v1                  # 可选，默认为 v1

基本认证（可选）

AIRFLOW_HOST=<your-airflow-host>        # 可选，默认为 http://localhost:8080
AIRFLOW_USERNAME=<your-airflow-username>
AIRFLOW_PASSWORD=<your-airflow-password>
AIRFLOW_API_VERSION=v1                  # 可选，默认为 v1

注意：如果提供了AIRFLOW_TOKEN，将使用它进行认证。否则，服务器将回退到使用用户名和密码的基本认证。

与Claude Desktop配合使用

首先，克隆仓库：

git clone https://github.com/nikhil-ganage/mcp-server-airflow-token

将以下内容添加到你的claude_desktop_config.json中：

使用令牌认证（推荐）

{
  "mcpServers": {
    "apache-airflow": {
      "type": "stdio",
      "command": "uv",
      "args": [
        "--directory",
        "path-to-repo/mcp-server-airflow-token",
        "run",
        "mcp-server-airflow-token"
      ],
      "env": {
        "AIRFLOW_HOST": "https://astro_id.astronomer.run/id",
        "AIRFLOW_TOKEN": "TOKEN"
      }
    }
  }
}

使用基本认证

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

对于只读模式（出于安全考虑推荐）：

只读模式下使用令牌认证

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

只读模式下使用基本认证

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

将path-to-repo替换为你克隆仓库的实际路径。

Astronomer Cloud配置示例

对于Astronomer Cloud部署：

{
  "mcpServers": {
    "mcp-server-airflow-token": {
      "command": "uvx",
      "args": ["mcp-server-airflow-token"],
      "env": {
        "AIRFLOW_HOST": "https://your-astronomer-domain.astronomer.run/your-deployment-id",
        "AIRFLOW_TOKEN": "your-astronomer-api-token"
      }
    }
  }
}

注意：部署ID是Astronomer Cloud URL路径的一部分。

选择API组

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

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

默认情况下使用所有API。允许的值包括：

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

只读模式

你可以使用--read-only标志以只读模式运行服务器。这将仅暴露执行读取操作（GET请求）的工具，并排除任何创建、更新或删除资源的工具。

uv run mcp-server-airflow-token --read-only

在只读模式下，服务器将仅暴露以下工具：

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

诸如创建、更新、删除DAG、变量、连接、触发DAG运行等写操作在只读模式下将不可用。

你可以将只读模式与API组选择结合使用：

uv run mcp-server-airflow-token --read-only --apis "dag,variable"

手动执行

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

make run

make run接受以下选项：

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

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

make run-sse

安装

你可以使用pip或uvx安装服务器：

# 使用pip
pip install mcp-server-airflow-token

# 使用uvx（推荐）
uvx mcp-server-airflow-token

🔧 开发

设置开发环境

克隆仓库：

git clone https://github.com/nikhil-ganage/mcp-server-airflow-token.git
cd mcp-server-airflow-token

安装开发依赖项：

uv sync --dev

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

touch .env

注意：运行测试不需要设置环境变量。出于开发和测试目的，AIRFLOW_HOST默认为http://localhost:8080。

运行测试

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

# 运行所有测试
make test

代码质量

# 运行代码检查
make lint

# 运行代码格式化
make format

持续集成

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

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

CI管道可确保在合并任何更改之前，代码质量和在支持的Python版本上的兼容性。

贡献

欢迎贡献代码！请随时提交拉取请求。

当pyproject.toml中的project.version更新时，该软件包将自动部署到PyPI。请遵循语义化版本控制。

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

📄 许可证

MIT许可证

mcp-server-airflow-token