MCP Airflow API

🚀 MCP-气流API

用自然语言管理Apache气流的革命性开源工具




______________________________________________________________________

建筑与内部（DeepWiki）


______________________________________________________________________

📋 概述

你有没有想过，如果你能使用自然语言而不是复杂的REST API调用或web界面操作来管理你的Apache Airflow工作流，那会有多神奇？ MCP-气流API 是一个革命性的开源项目，使这一目标成为现实。

MCP-Airflow-API Screenshot

______________________________________________________________________

🎯 什么是MCP-Airflow-API？

MCP-AirflowAPI是一个MCP服务器，它利用 模型上下文协议（MCP） 将Apache Airflow REST API操作转换为自然语言工具。该项目隐藏了API结构的复杂性，并通过自然语言命令实现了对Airflow集群的直观管理。

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

现在支持Airflow API v1（2.x）和v2（3.0+） 通过环境变量进行动态版本选择：

• API v1：与Airflow 2.x集群完全兼容（43个工具）- 文档
• API v2：Airflow 3.0+的增强功能，包括用于数据感知调度的资产管理（45个工具）- 文档

关键架构：具有共享通用工具（43）和v2专用资产工具（2）的单个MCP服务器-根据以下内容动态加载适当的工具集 AIRFLOW_API_VERSION 环境变量！

传统方法（示例）：

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

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

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

______________________________________________________________________

🚀 快速入门

📝 需要测试气流集群吗？ 使用我们的配套项目 在双方的支持下 气流2.x 和 气流3.x 环境！

快速入门/教程流程图

Flow Diagram of Quickstart/Tutorial

🎯 推荐：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. 转到“设置”→ 顶部菜单中的“工具”
4. 添加工具URL: http://localhost:8002/airflow-api
5. 配置您的LLM提供者（Ollama、OpenAI等）

______________________________________________________________________

📦 MCP服务器安装方法

方法1：从PyPI直接安装

uvx --python 3.12 mcp-airflow-api

方法2:Claude桌面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

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

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

远程访问（带承载令牌身份验证的流式http模式-推荐）

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

多个不同版本的气流集群

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

💡 专业建议：使用以下测试集群 对于上述配置，它们分别在端口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”

1. 全面的监控能力\

实时集群状态监控：

- 集群健康监测 - DAG状态和性能分析 - 任务执行日志跟踪 - XCom数据管理

1. 动态API版本支持\

单个MCP服务器可适应您的Airflow版本：

- API v1：43个用于Airflow 2.x兼容性的共享工具 - API v2：43个共享工具+2个Airflow 3.0资产管理工具+ - 环境变量控制：立即切换版本 AIRFLOW_API_VERSION - 零配置更改：工具名称相同，功能增强 - 高效架构：共享公共代码库消除了重复

1. 全面的工具覆盖\

涵盖几乎所有Airflow API功能：

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

1. 大环境优化\

高效处理1000多个DAGs的大型环境：

- 智能分页支持 - 高级过滤选项 - 批处理能力

______________________________________________________________________

🛠️ 技术优势

• 利用模型上下文协议（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)

• 全面气流API覆盖范围\

全面实施官方Airflow REST API：

- API v1支持：基于 气流2.x REST API - API v2支持：基于 气流3.0+REST API - 动态版本选择：API版本之间的运行时切换 - 特征奇偶校验：两个版本的完整端点覆盖

• 完整的Docker支持\

完整的Docker Compose设置，包含3个独立的服务：

- 打开WebUI：Web界面（端口 3002) - MCP服务器：气流API工具（内部端口 8000，通过暴露 18002) - MCPO代理：REST API端点提供程序（端口 8002)

______________________________________________________________________

实际使用案例

Capacity Management for Operations Teams

Capacity Management for Operations Teams

Capacity Management for Operations Teams

Capacity Management for Operations Teams

Capacity Management for Operations Teams

Capacity Management for Operations Teams

Capacity Management for Operations Teams

Capacity Management for Operations Teams

Capacity Management for Operations Teams

Capacity Management for Operations Teams

Capacity Management for Operations Teams

______________________________________________________________________

⚙️ 高级配置

环境变量

# 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: 气流2.x REST API参考
• API v2: 气流3.0+REST API参考

______________________________________________________________________

🔐 安全和身份验证

承载令牌身份验证

对于 streamable-http 模式下，此MCP服务器支持承载令牌身份验证，以确保远程访问的安全。在生产环境中运行服务器时，这一点尤为重要。

配置

启用身份验证：

# 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. 可流式传输http+REMOTE_AUTH_ENABLE=false：无身份验证的远程访问⚠️ 不建议用于生产
3. 可流式传输http+REMOTE_AUTH_ENABLE=true：使用承载令牌身份验证进行远程访问✅ 建议用于生产

备注: REMOTE_AUTH_ENABLE 默认为 false 当未定义、为空或为null时。支持的值为 true/false, 1/0, yes/no, on/off 不区分大小写

客户端配置

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

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

安全最佳实践

• 始终启用身份验证 在生产环境中使用流式http模式时
• 使用强随机生成的密钥 （建议使用32+个字符）
• 使用HTTPS 如果可能（使用SSL/TLS配置反向代理）
• 限制网络访问 使用防火墙或网络策略
• 定期旋转密钥 增强安全性
• 监控访问日志 未经授权的访问尝试

错误处理

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

• 401未经授权 用于丢失或无效的令牌
• 详细的错误消息 JSON格式用于调试

______________________________________________________________________

自定义Docker编写设置

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 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/* 端点
• 认证：基本身份验证
• 用例：生产就绪，功能稳定

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

用于测试 API v2 最新功能包括资产管理：

# 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/* 端点+资产管理
• 认证：JWT令牌（FabAuthManager）
• 用例：开发、测试新功能

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

对于不同气流版本的全面测试：

# 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

主要区别

清理

要停止并清理测试环境：

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

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

______________________________________________________________________

🌈 面向未来的架构

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

______________________________________________________________________

🎯 这个工具是给谁的？

• 数据工程师 --减少调试时间，提高生产率，最大限度地减少学习曲线
• DevOps工程师 --自动化基础设施监控，缩短事件响应时间
• 系统管理员 --用户友好的管理，无需复杂的API，实时监控集群状态

______________________________________________________________________

🚀 开源贡献和社区

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

如何做出贡献

• Bug报告和功能建议
• 文档改进
• 代码贡献

如果你觉得这个项目有用，请考虑主演。

______________________________________________________________________

🔮 结论

MCP-Airflow-API改变了数据工程和工作流管理的范式：\ 无需记住REST API调用-只需用自然语言询问：

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

______________________________________________________________________

🏷️ 标签

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

______________________________________________________________________

📚 示例查询和用例

本节提供了如何在自然语言查询中使用MCP-Airflow-API工具的全面示例。

基本DAG操作

• 列表_天：“以表格形式列出限制为10的所有DAG。”→ 最多可返回10天
• 列表_天：“以表格形式列出所有DAGs。”→ 最多返回所有DAG（警告：需要高令牌）
• 列表_天：“显示DAGs的下一页。”→ 使用偏移量进行分页
• 列表_天：“列出DAG21-40。”→ list_dags(limit=20, offset=20)
• 列表_天：“筛选ID包含‘tutorial’的DAGs。”→ list_dags(id_contains="etl")
• 列表_天：“筛选显示名称包含‘tutorial’的DAGs。”→ list_dags(name_contains="daily")
• get_dags_detailed_batch：“获取所有具有执行状态的DAG的详细信息。”→ get_dags_detailed_batch(fetch_all=True)
• get_dags_detailed_batch：“获取最近运行的活动、未使用的DAGs的详细信息。”→ get_dags_detailed_batch(is_active=True, is_paused=False)
• get_dags_detailed_batch：“获取包含'示例'和运行历史的DAGs的详细信息。”→ get_dags_detailed_batch(id_contains="example", limit=50)
• running_days：“显示正在运行的DAGs。”
• 失败_天：“显示失败的DAGs。”
• trigger_dag：“触发DAG'example_complex'。”
• pause_dag：“以表格式暂停DAG'example_complex'。”
• 取消DAG条款：“以表格式取消对DAG'example_complex'的暂停。”

集群管理和健康

• 获取健康：“检查气流群集运行状况。”
• 获取版本：“获取气流版本信息。”

游泳池管理

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

变量管理

• list_变量：“列出所有变量。”
• list_变量：“显示所有气流变量及其值。”
• 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天中状态为“失败”的任务实例。”
• get_task_instance_details：“获取DAG‘example_complex‘run‘scheduled\_\_xxxxx‘中任务‘data_processing‘的详细信息。”
• list_task_instances_batch：“列出上个月失败的任务实例。”
• list_task_instances_batch：“从本周开始批量显示多个DAG的任务实例。”
• get_task_instance_extra_links：“在最近一次运行中获取任务'data_processing'的额外链接。”
• get_task_instance_logs：“检索任务'create_entry_gcs'的日志，请尝试DAG'example_complex'的第2个。”

XCom管理

• list_xcom_entrys：“列出DAG‘example_complex‘run‘scheduled\_\_xxxxx‘中任务‘data_processing‘的XCom条目。”
• list_xcom_entrys：“显示最近运行中任务'data_processing'的所有XCom条目。”
• get_xcom_entry：“在特定运行中为任务'data_processing'获取带有键'result'的XCom条目。”
• get_xcom_entry：“从任务‘data_processing’中检索键‘processed_count’的XCom值。”

配置管理

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

重要：配置工具需要 expose_config = True 在airflow.cfg [webserver] 部分。如果禁用此功能，即使是管理员用户也会收到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())以及a 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_持续时间：“获取DAG'example_complex'的运行持续时间统计信息。”
• dag_ask_duration：“显示DAG'example_complex'的最新运行。”
• dag_ask_duration：“显示最近一次运行'manual\_\_xxxx'的任务持续时间。”
• dag_calendar：“获取DAG'example_complex'上个月的日历信息。”
• dag_calendar：“显示本周以来'example_complex'的DAG计划。”

日期计算示例

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

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

资产管理（仅限API v2版）

仅在以下情况下可用 AIRFLOW_API_VERSION=v2 （气流3.0+）：

• list_资产：“显示系统中注册的所有资产。”→ 列出用于数据感知调度的所有数据资产
• list_资产：“查找URI包含's3://数据湖'的资产。”→ 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创建的所有资产。”
• “查找最近未更新的资产。”
• “显示我们机器学习训练管道的数据沿袭。”

______________________________________________________________________

贡献

🤝 有想法吗？发现虫子了吗？想添加酷炫的功能吗？

我们总是很高兴欢迎新的贡献者！无论您是修复拼写错误、添加新的监控工具还是改进文档，每一份贡献都会使这个项目变得更好。

贡献方式：

• 🐛 报告问题或错误
• 💡 建议新的气流监测功能
• 📝 改进文档
• 🚀 提交拉取请求
• ⭐ 如果你觉得回购有用，请为其加星！

专业提示： 代码库的设计非常适合添加新工具。查看现有 @mcp.tool() 功能在 airflow_api.py.

______________________________________________________________________

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

此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许可证.