MCP Apache Airflow 服务管理
为 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
这是一个为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。