M

MCP Apache Airflow 服务管理

@yangkyeongmo/mcp-server-apache-airflow
Hosted
0 Stars 772 次浏览 yangkyeongmo 更新于 2026-08-23

为 MCP 客户端与 Apache Airflow 的 REST API 交互提供了一种标准化的方式,支持 DAG 管理和监控 Airflow 系统健康状态等操作。

MCP 服务配置

复制以下 JSON 到 OPClaw 或其他 MCP 客户端的配置文件中即可使用

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

该服务需要配置环境变量:AIRFLOW_HOST、AIRFLOW_PASSWORD、AIRFLOW_USERNAME

可用工具 (13 个)

该服务在 MCP 协议中暴露的工具,AI 可按需调用

list_dags 7 个参数

Lists all DAGs in the Airflow instance

该工具无需必填参数,直接调用即可

get_dag 1 个参数 需填 1 项

Get details of a specific DAG

必填参数:dag_id

pause_dag 1 个参数 需填 1 项

Pause a DAG

必填参数:dag_id

unpause_dag 1 个参数 需填 1 项

Unpause a DAG

必填参数:dag_id

trigger_dag 1 个参数 需填 1 项

Trigger a DAG run

必填参数:dag_id

get_dag_runs 13 个参数 需填 1 项

Get DAG runs for a specific DAG

必填参数:dag_id

get_dag_tasks 1 个参数 需填 1 项

Get tasks for a specific DAG

必填参数:dag_id

get_task_instance 3 个参数 需填 3 项

Get details of a specific task instance

必填参数:dag_id、task_id、dag_run_id

list_task_instances 17 个参数 需填 2 项

List all task instances for a specific DAG run

必填参数:dag_id、dag_run_id

get_import_error 1 个参数 需填 1 项

Get details of a specific import error

必填参数:import_error_id

list_import_errors 3 个参数

List all import errors

该工具无需必填参数,直接调用即可

get_health

Get the health status of the Airflow instance

该工具无需必填参数,直接调用即可

get_version

Get the version information of the Airflow instance

该工具无需必填参数,直接调用即可

服务介绍

mcp-server-apache-airflow

smithery badge

这是一个为Apache Airflow实现的模型上下文协议(MCP)服务器,能够与MCP客户端无缝集成。该项目提供了一种通过模型上下文协议与Apache Airflow交互的标准方法。

关于

本项目实现了一个模型上下文协议服务器,该服务器封装了Apache Airflow的REST API,允许MCP客户端以标准化方式与Airflow进行交互。它使用官方的Apache Airflow客户端库来确保兼容性和可维护性。

功能实现状态

Feature API Path Status
DAG Management
List DAGs /api/v1/dags
Get DAG Details /api/v1/dags/{dag_id}
Pause DAG /api/v1/dags/{dag_id}
Unpause DAG /api/v1/dags/{dag_id}
Update DAG /api/v1/dags/{dag_id}
Delete DAG /api/v1/dags/{dag_id}
Get DAG Source /api/v1/dagSources/{file_token}
Patch Multiple DAGs /api/v1/dags
Reparse DAG File /api/v1/dagSources/{file_token}/reparse
DAG Runs
List DAG Runs /api/v1/dags/{dag_id}/dagRuns
Create DAG Run /api/v1/dags/{dag_id}/dagRuns
Get DAG Run Details /api/v1/dags/{dag_id}/dagRuns/{dag_run_id}
Update DAG Run /api/v1/dags/{dag_id}/dagRuns/{dag_run_id}
Delete DAG Run /api/v1/dags/{dag_id}/dagRuns/{dag_run_id}
Get DAG Runs Batch /api/v1/dags/~/dagRuns/list
Clear DAG Run /api/v1/dags/{dag_id}/dagRuns/{dag_run_id}/clear
Set DAG Run Note /api/v1/dags/{dag_id}/dagRuns/{dag_run_id}/setNote
Get Upstream Dataset Events /api/v1/dags/{dag_id}/dagRuns/{dag_run_id}/upstreamDatasetEvents
Tasks
List DAG Tasks /api/v1/dags/{dag_id}/tasks
Get Task Details /api/v1/dags/{dag_id}/tasks/{task_id}
Get Task Instance /api/v1/dags/{dag_id}/dagRuns/{dag_run_id}/taskInstances/{task_id}
List Task Instances /api/v1/dags/{dag_id}/dagRuns/{dag_run_id}/taskInstances
Update Task Instance /api/v1/dags/{dag_id}/dagRuns/{dag_run_id}/taskInstances/{task_id}
Clear Task Instances /api/v1/dags/{dag_id}/clearTaskInstances
Set Task Instances State /api/v1/dags/{dag_id}/updateTaskInstancesState
Variables
List Variables /api/v1/variables
Create Variable /api/v1/variables
Get Variable /api/v1/variables/{variable_key}
Update Variable /api/v1/variables/{variable_key}
Delete Variable /api/v1/variables/{variable_key}
Connections
List Connections /api/v1/connections
Create Connection /api/v1/connections
Get Connection /api/v1/connections/{connection_id}
Update Connection /api/v1/connections/{connection_id}
Delete Connection /api/v1/connections/{connection_id}
Test Connection /api/v1/connections/test
Pools
List Pools /api/v1/pools
Create Pool /api/v1/pools
Get Pool /api/v1/pools/{pool_name}
Update Pool /api/v1/pools/{pool_name}
Delete Pool /api/v1/pools/{pool_name}
XComs
List XComs /api/v1/dags/{dag_id}/dagRuns/{dag_run_id}/taskInstances/{task_id}/xcomEntries
Get XCom Entry /api/v1/dags/{dag_id}/dagRuns/{dag_run_id}/taskInstances/{task_id}/xcomEntries/{xcom_key}
Datasets
List Datasets /api/v1/datasets
Get Dataset /api/v1/datasets/{uri}
Get Dataset Events /api/v1/datasetEvents
Create Dataset Event /api/v1/datasetEvents
Get DAG Dataset Queued Event /api/v1/dags/{dag_id}/dagRuns/queued/datasetEvents/{uri}
Get DAG Dataset Queued Events /api/v1/dags/{dag_id}/dagRuns/queued/datasetEvents
Delete DAG Dataset Queued Event /api/v1/dags/{dag_id}/dagRuns/queued/datasetEvents/{uri}
Delete DAG Dataset Queued Events /api/v1/dags/{dag_id}/dagRuns/queued/datasetEvents
Get Dataset Queued Events /api/v1/datasets/{uri}/dagRuns/queued/datasetEvents
Delete Dataset Queued Events /api/v1/datasets/{uri}/dagRuns/queued/datasetEvents
Monitoring
Get Health /api/v1/health
DAG Stats
Get DAG Stats /api/v1/dags/statistics
Config
Get Config /api/v1/config
Plugins
Get Plugins /api/v1/plugins
Providers
List Providers /api/v1/providers
Event Logs
List Event Logs /api/v1/eventLogs
Get Event Log /api/v1/eventLogs/{event_log_id}
System
Get Import Errors /api/v1/importErrors
Get Import Error Details /api/v1/importErrors/{import_error_id}
Get Health Status /api/v1/health
Get Version /api/v1/version

设置

依赖项

此项目依赖于官方的 Apache Airflow 客户端库 (apache-airflow-client)。当你安装此包时,它将被自动安装。

环境变量

设置以下环境变量:

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

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

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

/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 Server:

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

贡献

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

许可证

MIT 许可证

相关 MCP 服务