Mcp Airflow Api

🚀 MCP-Airflow-API

革新性的开源工具，助你通过自然语言管理 Apache Airflow

---

架构与内部机制（DeepWiki）

---

📋 概述

你是否想过，如果能用自然语言管理 Apache Airflow 工作流，而不必处理复杂的 REST API 调用或操作 Web 界面，那会有多么出色？MCP-Airflow-API 是一个革命性的开源项目，让这一目标成为现实。

---

🎯 MCP-Airflow-API 是什么？

MCP-Airflow-API 是一个 MCP 服务，它利用 Model Context Protocol（MCP） 将 Apache Airflow REST API 操作转换为自然语言工具。该项目隐藏了 API 结构的复杂性，让你可以通过自然语言命令直观地管理 Airflow 集群。

🆕 多版本 API 支持（新增！）

现已同时支持 Airflow API v1（2.x）和 v2（3.0+），并可通过环境变量动态选择版本：

• API v1：全面兼容 Airflow 2.x 集群（54 个工具）——文档
• API v2：为 Airflow 3.0+ 提供增强功能，包括用于数据感知调度的资产管理（已注册 56 个工具）——文档

核心架构：单个 MCP 服务包含共享的通用工具（54 个）和 v2 专属资产工具（2 个），并根据 AIRFLOW_API_VERSION 环境变量动态加载相应的工具集！

传统方式（示例）：

curl -X GET "http://localhost:8080/api/v1/dags?limit=100&offset=0" \
  -H "Authorization: Basic YWlyZmxvdzphaXJmbG93"

MCP-Airflow-API 方式（自然语言）：

“显示当前正在运行的 DAG”

---

🚀 快速入门

📝 需要用于测试的 Airflow 集群？ 使用我们的配套项目 Airflow-Docker-Compose，该项目同时支持 Airflow 2.x 和 Airflow 3.x 环境！

快速入门／教程流程图

🎯 推荐：Docker Compose（完整演示环境）

适用于快速评估和测试：

git clone https://github.com/call518/MCP-Airflow-API.git
cd MCP-Airflow-API

# Configure your Airflow credentials
cp .env.example .env
# Edit .env with your Airflow API settings

# Start all services
docker-compose up -d

# Access OpenWebUI at http://localhost:3002/
# API documentation at http://localhost:8002/docs

使用 OpenWebUI 入门（Docker 方式）

📌 注意：Web UI 配置说明基于 OpenWebUI v0.6.22。较新版本中的菜单位置和设置可能有所不同。

1. 访问 http://localhost:3002/
2. 使用管理员账号登录
3. 在顶部菜单中前往“Settings” → “Tools”
4. 添加工具 URL：http://localhost:8002/airflow-api
5. 配置你的大模型提供方（Ollama、OpenAI 等）

---

📦 MCP 服务安装方式

方法 1：直接从 PyPI 安装

uvx --python 3.12 mcp-airflow-api

方法 2：Claude-Desktop MCP 客户端集成

本地访问（stdio 模式）

{
  "mcpServers": {
    "mcp-airflow-api": {
      "command": "uvx",
      "args": ["--python", "3.12", "mcp-airflow-api"],
      "env": {
        "AIRFLOW_API_VERSION": "v2",
        "AIRFLOW_API_BASE_URL": "http://localhost:8080/api",
        "AIRFLOW_API_USERNAME": "airflow",
        "AIRFLOW_API_PASSWORD": "airflow"
      }
    }
  }
}git

远程访问（无身份验证的 streamable-http 模式）

{
  "mcpServers": {
    "mcp-airflow-api": {
      "type": "streamable-http",
      "url": "http://localhost:8000/mcp"
    }
  }
}

远程访问（使用 Bearer 令牌身份验证的 streamable-http 模式——推荐）

{
  "mcpServers": {
    "mcp-airflow-api": {
      "type": "streamable-http",
      "url": "http://localhost:8000/mcp",
      "headers": {
        "Authorization": "Bearer your-secure-secret-key-here"
      }
    }
  }
}

使用不同版本的多个 Airflow 集群

{
  "mcpServers": {
    "airflow-2x-cluster": {
      "command": "uvx",
      "args": ["--python", "3.12", "mcp-airflow-api"],
      "env": {
        "AIRFLOW_API_VERSION": "v1",
        "AIRFLOW_API_BASE_URL": "http://localhost:38080/api",
        "AIRFLOW_API_USERNAME": "airflow",
        "AIRFLOW_API_PASSWORD": "airflow"
      }
    },
    "airflow-3x-cluster": {
      "command": "uvx",
      "args": ["--python", "3.12", "mcp-airflow-api"],
      "env": {
        "AIRFLOW_API_VERSION": "v2",
        "AIRFLOW_API_BASE_URL": "http://localhost:48080/api",
        "AIRFLOW_API_USERNAME": "airflow",
        "AIRFLOW_API_PASSWORD": "airflow"
      }
    }
  }
}

💡 专业提示：上述配置可使用 Airflow-Docker-Compose 中的测试集群——它们分别运行在端口 38080（2.x）和 48080（3.x）上！

方法 3：开发环境安装

git clone https://github.com/call518/MCP-Airflow-API.git
cd MCP-Airflow-API
pip install -e .

# Run in stdio mode
python -m mcp_airflow_api

---

🌟 主要功能

1. 自然语言查询
   无需学习复杂的 API 语法。像平常说话一样提问即可：

   • “目前有哪些 DAG 正在运行？”
   • “显示失败的任务”
   • “查找包含 ETL 的 DAG”

2. 全面的监控能力
   实时监控集群状态：

   • 集群运行状况监控
   • DAG 状态与性能分析
   • 任务执行日志跟踪
   • XCom 数据管理

3. 动态 API 版本支持
   单个 MCP 服务可适配你的 Airflow 版本：

   • API v1：54 个共享工具，兼容 Airflow 2.x
   • API v2：54 个共享工具 + 2 个资产管理工具，适用于 Airflow 3.0+
   • 环境变量控制：通过 AIRFLOW_API_VERSION 即时切换版本
   • 无需更改配置：工具名称保持不变，功能得到增强
   • 高效架构：共享通用代码库，避免重复

4. 全面的工具覆盖
   覆盖几乎所有 Airflow API 功能：

   • DAG 管理（触发、暂停、恢复）
   • 任务实例监控
   • Pool 和变量管理
   • Connection 配置
   • 配置查询
   • 事件日志分析

5. 大型环境优化
   可高效处理包含 1000 多个 DAG 的大型环境：

   • 智能分页支持
   • 高级筛选选项
   • 批处理能力

---

🛠️ 技术优势

• 利用 Model Context Protocol (MCP)
  MCP 是一种开放标准，用于在 AI 应用与数据源之间建立安全连接，并提供：

  • 标准化接口
  • 安全的数据访问
  • 可扩展架构

• 支持两种传输模式

  • stdio 模式：适用于本地环境的直接 MCP 客户端集成
  • streamable-http 模式：基于 HTTP 的部署，适用于 Docker 和远程访问

  环境变量控制：

  FASTMCP_TYPE=stdio          # Default: Direct MCP client mode
  FASTMCP_TYPE=streamable-http # Docker/HTTP mode
  FASTMCP_PORT=8000           # HTTP server port (Docker internal)
  

• 全面覆盖 Airflow API
  完整实现官方 Airflow REST API：

  • API v1 支持：基于 Airflow 2.x REST API
  • API v2 支持：基于 Airflow 3.0+ REST API
  • 动态版本选择：在运行时切换 API 版本
  • 功能对等：两个版本均完整覆盖所有端点

• 完整的 Docker 支持
  完整的 Docker Compose 配置，包含 3 个独立服务：

  • Open WebUI：Web 界面（端口 3002）
  • MCP 服务：Airflow API 工具（内部端口 8000，通过 18002 暴露）
  • MCPO Proxy：REST API 端点提供程序（端口 8002）

---

实际用例

---

⚙️ 高级配置

环境变量

# Required - Dynamic API Version Selection (NEW!)
# Single server supports both v1 and v2 - just change this variable!
AIRFLOW_API_VERSION=v1           # v1 for Airflow 2.x, v2 for Airflow 3.0+
AIRFLOW_API_BASE_URL=http://localhost:8080/api

# Test Cluster Connection Examples:
# For Airflow 2.x test cluster (from Airflow-Docker-Compose)
AIRFLOW_API_VERSION=v1
AIRFLOW_API_BASE_URL=http://localhost:38080/api

# For Airflow 3.x test cluster (from Airflow-Docker-Compose)  
AIRFLOW_API_VERSION=v2
AIRFLOW_API_BASE_URL=http://localhost:48080/api

# Authentication
AIRFLOW_API_USERNAME=airflow
AIRFLOW_API_PASSWORD=airflow

# Optional - MCP Server Configuration
MCP_LOG_LEVEL=INFO                   # DEBUG/INFO/WARNING/ERROR/CRITICAL
FASTMCP_TYPE=stdio                   # stdio/streamable-http
FASTMCP_PORT=8000                    # HTTP server port (Docker mode)

# Bearer Token Authentication for streamable-http mode
# Enable authentication (recommended for production)
# Default: false (when undefined, empty, or null)
# Values: true/false, 1/0, yes/no, on/off (case insensitive)
REMOTE_AUTH_ENABLE=false             # true/false
REMOTE_SECRET_KEY=your-secure-secret-key-here

API 版本对比

官方文档：

• API v1：Airflow 2.x REST API 参考
• API v2：Airflow 3.0+ REST API 参考

功能	API v1（Airflow 2.x）	API v2（Airflow 3.0+）
工具总数	54 个工具	56 个已注册工具
共用工具	54（100%）	54（96%；包含 8 个仅限 v1 的工具）
独有工具	0	2（资源管理）
基础 DAG 操作	✅	✅ 增强
任务管理	✅	✅ 增强
连接管理	✅	✅ 增强
池管理	✅	✅ 增强
资源管理	❌	✅ 新增
资源事件	❌	✅ 新增
数据感知调度	❌	✅ 新增
增强的 DAG 警告	❌	✅ 新增
高级筛选	基础	✅ 增强

计数指已注册的 MCP 工具。在 API v2 中，八个已注册工具 （list_users、get_user、list_permissions、list_roles、list_datasets、 get_dataset、list_dataset_events 和 get_dataset_events）会返回 不支持的操作响应。使用 list_assets 和 list_asset_events 代替 v2 中的数据集工具。

---

🔐 安全与身份验证

Bearer 令牌身份验证

在 streamable-http 模式下，此 MCP 服务支持 Bearer 令牌身份验证，以保护远程访问。这在生产环境中运行服务时尤其重要。

配置

启用身份验证：

# In .env file
REMOTE_AUTH_ENABLE=true
REMOTE_SECRET_KEY=your-secure-secret-key-here

或通过 CLI：

python -m mcp_airflow_api --type streamable-http --auth-enable --secret-key your-secure-secret-key-here

安全级别

1. stdio 模式（默认）：仅限本地访问，无需身份验证
2. streamable-http + REMOTE_AUTH_ENABLE=false：远程访问，不进行身份验证 ⚠️ 不建议用于生产环境
3. streamable-http + REMOTE_AUTH_ENABLE=true：远程访问，使用 Bearer 令牌身份验证 ✅ 建议用于生产环境

注意：REMOTE_AUTH_ENABLE 在未定义、为空或为 null 时默认为 false。支持的值包括 true/false、1/0、yes/no、on/off（不区分大小写）。

客户端配置

启用身份验证后，MCP 客户端必须在 Authorization 标头中包含 Bearer 令牌：

{
  "mcpServers": {
    "mcp-airflow-api": {
      "type": "streamable-http",
      "url": "http://your-server:8000/mcp",
      "headers": {
        "Authorization": "Bearer your-secure-secret-key-here"
      }
    }
  }
}

安全最佳实践

• 在生产环境中使用 streamable-http 模式时，务必启用身份验证
• 使用强随机生成的密钥（建议长度为 32 个字符以上）
• 尽可能使用 HTTPS（配置带有 SSL/TLS 的反向代理）
• 使用防火墙或网络策略限制网络访问
• 定期轮换密钥以增强安全性
• 监控访问日志，检查未经授权的访问尝试

错误处理

身份验证失败时，服务会返回：

• 401 Unauthorized（缺少或令牌无效）
• 用于调试的 JSON 格式详细错误消息

---

自定义 Docker Compose 设置

version: '3.8'
services:
  mcp-server:
    build: 
      context: .
      dockerfile: Dockerfile.MCP-Server
    environment:
      - FASTMCP_PORT=8000
      - AIRFLOW_API_VERSION=v1
      - AIRFLOW_API_BASE_URL=http://your-airflow:8080/api
      - AIRFLOW_API_USERNAME=airflow
      - AIRFLOW_API_PASSWORD=airflow

开发安装

git clone https://github.com/call518/MCP-Airflow-API.git
cd MCP-Airflow-API
pip install -e .

# Run in stdio mode
python -m mcp_airflow_api

---

🧪 测试 Airflow 集群部署

在测试和开发中，请使用我们的配套项目 Airflow-Docker-Compose，该项目同时支持 Airflow 2.x 和 3.x 环境。

快速设置

1. 克隆测试环境仓库：

   git clone https://github.com/call518/Airflow-Docker-Compose.git
   cd Airflow-Docker-Compose
   

选项 1：部署 Airflow 2.x（LTS）

用于测试 API v1 兼容性及稳定的生产功能：

# Navigate to Airflow 2.x environment
cd airflow-2.x

# (Optional) Customize environment variables
cp .env.template .env
# Edit .env file as needed

# Deploy Airflow 2.x cluster
./run-airflow-cluster.sh

# Access Web UI
# URL: http://localhost:38080
# Username: airflow / Password: airflow

环境详情：

• 镜像：apache/airflow:2.10.2
• 端口：38080（可通过 AIRFLOW_WEBSERVER_PORT 配置）
• API：/api/v1/* 端点
• 身份验证：Basic Auth
• 适用场景：生产就绪的稳定功能

选项 2：部署 Airflow 3.x（最新版本）

用于测试 API v2 和包括 Assets 管理在内的最新功能：

# Navigate to Airflow 3.x environment  
cd airflow-3.x

# (Optional) Customize environment variables
cp .env.template .env
# Edit .env file as needed

# Deploy Airflow 3.x cluster
./run-airflow-cluster.sh

# Access API Server
# URL: http://localhost:48080
# Username: airflow / Password: airflow

环境详情：

• 镜像：apache/airflow:3.0.6
• 端口：48080（可通过 AIRFLOW_APISERVER_PORT 配置）
• API：/api/v2/* 端点 + Assets 管理
• 身份验证：JWT Token（FabAuthManager）
• 适用场景：开发和测试新功能

选项 3：同时部署两个版本

用于在不同 Airflow 版本间进行全面测试：

# Start Airflow 2.x (port 38080)
cd airflow-2.x && ./run-airflow-cluster.sh

# Start Airflow 3.x (port 48080) 
cd ../airflow-3.x && ./run-airflow-cluster.sh

主要区别

功能	Airflow 2.x	Airflow 3.x
身份验证	Basic Auth	JWT Tokens（FabAuthManager）
默认端口	38080	48080
API 端点	/api/v1/*	/api/v2/*
Assets 支持	❌ 有限／实验性	✅ 完整支持
Provider 包	providers	distributions
稳定性	✅ 可用于生产	🧪 Beta／开发中

清理

停止并清理测试环境：

# For Airflow 2.x
cd airflow-2.x && ./cleanup-airflow-cluster.sh

# For Airflow 3.x
cd airflow-3.x && ./cleanup-airflow-cluster.sh

---

🌈 面向未来的架构

• 可扩展的设计和模块化结构，便于添加新功能
• 符合标准的协议，可与其他工具集成
• 云原生运维和适配大模型的接口
• 具备上下文感知查询处理和自动化工作流管理能力

---

🎯 这个工具适合哪些人？

• 数据工程师 — 缩短调试时间、提高工作效率并降低学习门槛
• DevOps 工程师 — 自动化基础设施监控，缩短事件响应时间
• 系统管理员 — 无需处理复杂 API，即可通过易用的界面管理并实时监控集群状态

---

🚀 开源贡献与社区

仓库： https://github.com/call518/MCP-Airflow-API

如何贡献

• 报告错误并提出功能建议
• 改进文档
• 贡献代码

如果你觉得这个项目有用，请考虑为项目点亮星标。

---

🔮 总结

MCP-Airflow-API 改变了数据工程和工作流管理的模式：
无需记忆 REST API 调用，只需用自然语言提问：

“显示当前正在运行的 ETL 作业状态。”

---

🏷️ 标签

#Apache-Airflow #MCP #ModelContextProtocol #DataEngineering #DevOps #WorkflowAutomation #NaturalLanguage #OpenSource #Python #Docker #AI-Integration

---

📚 示例查询与使用场景

本节全面介绍如何使用自然语言查询调用 MCP-Airflow-API 工具。

DAG 基本操作

• list_dags：“以表格形式列出最多 10 个 DAG。” → 最多返回 10 个 DAG
• list_dags：“以表格形式列出所有 DAG。” → 返回全部 DAG（警告：需要大量 Token）
• list_dags：“显示下一页 DAG。” → 使用 offset 进行分页
• list_dags：“列出第 21–40 个 DAG。” → list_dags(limit=20, offset=20)
• list_dags：“筛选 ID 包含 ‘tutorial’ 的 DAG。” → list_dags(id_contains="etl")
• list_dags：“筛选显示名称包含 ‘tutorial’ 的 DAG。” → list_dags(name_contains="daily")
• get_dags_detailed_batch：“获取所有 DAG 的详细信息和执行状态。” → get_dags_detailed_batch(fetch_all=True)
• get_dags_detailed_batch：“获取活跃、未暂停且有近期运行记录的 DAG 的详细信息。” → get_dags_detailed_batch(is_active=True, is_paused=False)
• get_dags_detailed_batch：“获取包含 ‘example’ 的 DAG 的详细信息及运行历史。” → get_dags_detailed_batch(id_contains="example", limit=50)
• running_dags：“显示正在运行的 DAG。”
• failed_dags：“显示运行失败的 DAG。”
• trigger_dag：“触发 DAG ‘example_complex’。”
• pause_dag：“以表格形式暂停 DAG ‘example_complex’。”
• unpause_dag：“以表格形式取消暂停 DAG ‘example_complex’。”

集群管理与健康状况

• get_health：“检查 Airflow 集群健康状况。”
• get_version：“获取 Airflow 版本信息。”

池管理

• list_pools：“列出所有池。”
• list_pools：“显示池使用统计信息。”
• get_pool：“获取池 ‘default_pool’ 的详细信息。”
• get_pool：“检查池利用率。”

变量管理

• list_variables：“列出所有变量。”
• list_variables：“显示所有 Airflow 变量及其值。”
• get_variable：“获取变量 ‘database_url’。”
• get_variable：“显示变量 ‘api_key’ 的值。”

任务实例管理

• list_task_instances_all：“列出 DAG ‘example_complex’ 的所有任务实例。”
• list_task_instances_all：“显示正在运行的任务实例。”
• list_task_instances_all：“显示按池 ‘default_pool’ 筛选的任务实例。”
• list_task_instances_all：“列出运行时长超过 300 秒的任务实例。”
• list_task_instances_all：“显示上周运行失败的任务实例。”
• list_task_instances_all：“列出昨天运行失败的任务实例。”
• list_task_instances_all：“显示今天上午 9 点后启动的任务实例。”
• list_task_instances_all：“列出过去 3 天状态为 ‘failed’ 的任务实例。”
• get_task_instance_details：“获取 DAG ‘example_complex’ 的运行记录 ‘scheduled__xxxxx’ 中任务 ‘data_processing’ 的详细信息。”
• list_task_instances_batch：“列出上个月运行失败的任务实例。”
• list_task_instances_batch：“显示本周多个 DAG 的批量任务实例。”
• get_task_instance_extra_links：“获取最近一次运行中任务 ‘data_processing’ 的额外链接。”
• get_task_instance_logs：“获取 DAG ‘example_complex’ 中任务 ‘create_entry_gcs’ 第 2 次尝试的日志。”

XCom 管理

• list_xcom_entries：“列出 DAG ‘example_complex’ 的运行记录 ‘scheduled__xxxxx’ 中任务 ‘data_processing’ 的 XCom 条目。”
• list_xcom_entries：“显示最近一次运行中任务 ‘data_processing’ 的所有 XCom 条目。”
• get_xcom_entry：“获取特定运行中任务 ‘data_processing’ 的键为 ‘result’ 的 XCom 条目。”
• get_xcom_entry：“获取任务 ‘data_processing’ 中键为 ‘processed_count’ 的 XCom 值。”

配置管理

• get_config：“显示所有 Airflow 配置分区和选项。” → 返回完整配置；如果 expose_config=False，则返回 403
• list_config_sections：“列出所有配置分区及其摘要信息。”
• get_config_section：“获取 core 分区中的所有设置。” → get_config_section("core")
• get_config_section：“显示 Web 服务器配置选项。” → get_config_section("webserver")
• search_config_options：“查找所有与数据库相关的配置选项。” → search_config_options("database")
• search_config_options：“搜索配置中的超时设置。” → search_config_options("timeout")

重要：配置工具要求在 airflow.cfg 的 [webserver] 分区中设置 expose_config = True。即使是管理员用户，在此选项禁用时也会收到 403 错误。

DAG 分析与监控

• get_dag：“获取 DAG 'example_complex' 的详细信息。”
• get_dags_detailed_batch：“获取所有 DAG 的综合详细信息，包括执行历史记录。” → get_dags_detailed_batch(fetch_all=True)
• get_dags_detailed_batch：“获取活跃 DAG 的详细信息及最新运行信息。” → get_dags_detailed_batch(is_active=True)
• get_dags_detailed_batch：“获取 ETL DAG 的详细信息及近期执行数据。” → get_dags_detailed_batch(id_contains="etl")

注意：get_dags_detailed_batch 返回每个 DAG 的配置详细信息（来自 get_dag()），以及一个 latest_dag_run 字段，其中包含最近一次执行的信息（run_id、state、execution_date、start_date、end_date 等）。

• dag_graph：“显示 DAG 'example_complex' 的任务图。”
• list_tasks：“列出 DAG 'example_complex' 中的所有任务。”
• dag_code：“获取 DAG 'example_complex' 的源代码。”
• list_event_logs：“列出 DAG 'example_complex' 的事件日志。”
• list_event_logs：“显示所有 DAG 中 ID 为昨天的事件日志。”
• get_event_log：“获取 ID 为 12345 的事件日志条目。”
• all_dag_event_summary：“显示所有 DAG 的事件计数摘要。”
• list_import_errors：“列出带有 ID 的导入错误。”
• get_import_error：“获取 ID 为 67890 的导入错误。”
• all_dag_import_summary：“显示所有 DAG 的导入错误摘要。”
• dag_run_duration：“获取 DAG 'example_complex' 的运行时长统计信息。”
• dag_task_duration：“显示 DAG 'example_complex' 的最新运行。”
• dag_task_duration：“显示 'manual__xxxxx' 最新运行的任务时长。”
• dag_calendar：“获取 DAG 'example_complex' 上个月的日历信息。”
• dag_calendar：“显示 DAG 'example_complex' 本周的调度计划。”

其他已注册工具

类别	工具	用途	API 支持
提示词模板	get_prompt_template	读取模板，或使用 section 选择一个部分	v1, v2
连接	list_connections	使用 limit 和 offset 列出连接	v1, v2
连接	get_connection	根据 connection_id 读取连接	v1, v2
连接	create_connection	根据 connection_data 创建连接	v1, v2
连接	update_connection	使用 connection_data 更新 connection_id	v1, v2
连接	delete_connection	根据 connection_id 删除连接	v1, v2
用户和权限	list_users	使用 limit 和 offset 列出用户	仅 v1
用户和权限	get_user	根据 username 读取用户	仅 v1
用户和权限	list_permissions	列出可用权限	仅 v1
用户和权限	list_roles	使用 limit 和 offset 列出角色	仅 v1
插件和提供程序	list_plugins	列出已安装的插件	v1, v2
插件和提供程序	list_providers	列出已安装的提供程序包	v1, v2
插件和提供程序	get_provider	根据 provider_name 读取提供程序	v1, v2
数据集	list_datasets	列出数据集，可按 uri_pattern 筛选	仅 v1
数据集	get_dataset	根据 dataset_uri 读取数据集	仅 v1
数据集	list_dataset_events	按 dataset_uri 或 source_dag_id 筛选事件	仅 v1
数据集	get_dataset_events	列出特定 dataset_uri 的事件	仅 v1

日期计算示例

工具会自动根据服务当前的日期/时间进行相对日期计算：

用户输入	计算方法	示例格式
“昨天”	current_date - 1 day	YYYY-MM-DD（当前日期前 1 天）
“上周”	current_date - 7 days to current_date - 1 day	YYYY-MM-DD 到 YYYY-MM-DD（7 天范围）
“过去 3 天”	current_date - 3 days to current_date	YYYY-MM-DD 到 YYYY-MM-DD（3 天范围）
“今天早上”	current_date 00:00 to 12:00	YYYY-MM-DDTHH:mm:ssZ 格式

服务始终使用当前日期/时间进行这些计算。

资产管理（仅限 API v2）

仅当 AIRFLOW_API_VERSION=v2（Airflow 3.0+）时可用：

• list_assets：“显示系统中注册的所有资产。” → 列出所有用于数据感知调度的数据资产
• list_assets：“查找 URI 包含 's3://data-lake' 的资产。” → list_assets(uri_pattern="s3://data-lake")
• list_asset_events：“显示最近的资产事件。” → 列出资产的创建或更新时间
• list_asset_events：“显示特定 URI 的资产事件。” → list_asset_events(asset_uri="s3://bucket/file.csv")
• list_asset_events：“查找 ETL DAG 产生的事件。” → list_asset_events(source_dag_id="etl_pipeline")

数据感知调度示例：

• “告诉我哪些资产会触发 customer_analysis DAG。”
• “列出本周由 data_ingestion DAG 创建的所有资产。”
• “查找最近没有更新的资产。”
• “显示 ML 训练流水线的数据沿袭。”

---

参与贡献

🤝 有想法？发现 Bug？想添加酷炫的功能？

我们始终欢迎新的贡献者！无论你是在修正错别字、添加新的监控工具，还是改进文档，每一份贡献都会让这个项目变得更好。

参与贡献的方式：

• 🐛 报告问题或 Bug
• 💡 提出新的 Airflow 监控功能
• 📝 改进文档
• 🚀 提交拉取请求
• ⭐ 如果觉得这个仓库有用，就给它点个 Star！

小贴士： 代码库的设计非常便于添加新工具。请查看 airflow_api.py 中现有的 @mcp.tool() 函数。

---

🛠️ 添加自定义工具（高级）

此 MCP 服务旨在方便扩展。了解主要功能并完成快速入门后，你可以按以下步骤添加自己的自定义工具：

分步指南

1. 添加辅助函数（可选）

在 src/mcp_airflow_api/functions.py 中添加可复用的数据函数：

async def get_your_custom_data(target_resource: str = None) -> List[Dict[str, Any]]:
  """Your custom data retrieval function."""
  # Example implementation - adapt to your service
  data_source = await get_data_connection(target_resource)
  results = await fetch_data_from_source(
    source=data_source,
    filters=your_conditions,
    aggregations=["count", "sum", "avg"],
    sorting=["count DESC", "timestamp ASC"]
  )
  return results

2. 创建 MCP 工具

在 src/mcp_airflow_api/airflow_api.py 中添加你的工具函数：

@mcp.tool()
async def get_your_custom_analysis(limit: int = 50, target_name: Optional[str] = None) -> str:
  """
  [Tool Purpose]: Brief description of what your tool does
    
  [Exact Functionality]:
  - Feature 1: Data aggregation and analysis
  - Feature 2: Resource monitoring and insights
  - Feature 3: Performance metrics and reporting
    
  [Required Use Cases]:
  - When user asks "your specific analysis request"
  - Your business-specific monitoring needs
    
  Args:
    limit: Maximum results (1-100)
    target_name: Target resource/service name
    
  Returns:
    Formatted analysis results
  """
  try:
    limit = max(1, min(limit, 100))  # Always validate input
    results = await get_your_custom_data(target_resource=target_name)
    if results:
      results = results[:limit]
    return format_table_data(results, f"Custom Analysis (Top {len(results)})")
  except Exception as e:
    logger.error(f"Failed to get custom analysis: {e}")
    return f"Error: {str(e)}"

3. 更新导入项（如有需要）

在 src/mcp_airflow_api/airflow_api.py 的导入项中添加你的辅助函数：

from .functions import (
  # ...existing imports...
  get_your_custom_data,  # Add your new function
)

4. 更新提示词模板（推荐）

在 src/mcp_airflow_api/prompt_template.md 中添加你的工具描述，以便更好地识别自然语言：

### **Your Custom Analysis Tool**

### X. **get_your_custom_analysis**
**Purpose**: Brief description of what your tool does
**Usage**: "Show me your custom analysis" or "Get custom analysis for database_name"
**Features**: Data aggregation, resource monitoring, performance metrics
**Required**: `target_name` parameter for specific resource analysis

5. 测试你的工具

# Local testing
./scripts/run-mcp-inspector-local.sh

# Or with Docker
docker-compose up -d
docker-compose logs -f mcp-server

# Test with natural language:
# "Show me your custom analysis"
# "Get custom analysis for target_name"

就是这么简单！你的自定义工具现已可以通过自然语言查询使用。

许可证

你可以在 MIT License 下自由使用、修改和分发。