Apache Airflow与MCP服务器
使用Docker Compose和集成MCP（模型上下文协议）服务器进行完整的Apache Airflow部署，通过Warp终端进行AI驱动的工作流管理。
🚀 快速开始
# Navigate to project directory
cd airflow-deploy-mcp
# Start all services
./start.sh
# Access Airflow UI at http://localhost:8080
# Username: admin
# Password: admin123✨ 特性
气流组件
- PostgreSQL：元数据数据库
- 瑞迪斯：Celery的消息代理
- 气流Web服务器：Web UI（端口8080）
- 气流调度器：DAG调度和编排
- 气流工人：用于任务执行的Celery执行器
- 气流触发器：支持可延期运营商
MCP服务器集成
- 🤖 人工智能集成：通过Warp终端中的AI助手控制气流
- 📊 DAG管理：列出、触发和监视DAGs
- 📈 实时监控：检查DAG运行状态和任务日志
- 🔧 配置：管理气流变量和连接
- 🔐 安全：通过Airflow REST API进行身份验证访问
可用的MCP工具
DAG管理
list_dags-列出所有可用的DAGsget_dag-获取详细的DAG信息trigger_dag-触发器DAG随配置运行pause_dag-暂停DAG以防止新运行unpause_dag-取消DAG的暂停以允许调度get_dag_runs-查看DAG运行历史记录get_dag_run_status-检查特定运行状态
DAG文件管理
list_dag_files-列出dags文件夹中的所有DAG文件read_dag_file-读取DAG文件的内容upload_dag_file-上传新的DAG或更新现有的DAGdelete_dag_file-删除DAG文件validate_dag_file-验证Python语法和DAG结构
任务管理
get_task_instances-列出DAG运行中的任务get_task_logs-检索任务执行日志
配置
list_variables-列出气流变量get_variable-获取变量值set_variable-设置或更新变量delete_variable-删除变量list_connections-列出气流连接
📋 先决条件
- Docker 桌面版：版本20.10或更高版本
- Docker Compose：2.0或更高版本
- 记忆：建议最低4GB RAM
- 磁盘空间：至少5GB可用空间
- 经纱终端：用于MCP集成（可选）
🛠️ 安装
1.克隆或导航到项目
cd airflow-deploy-mcp2.配置环境变量
查看并自定义 .env 文件：
cat .env重要：更改生产环境的默认密码！
3.启动服务
使用提供的脚本：
./start.sh或者手动使用Docker Compose：
docker-compose up -d首次启动需要2-3分钟：
- 下载Docker镜像
- 初始化气流数据库
- 创建管理员用户
- 启动所有服务
4.验证状态
docker-compose ps所有服务都应显示为“健康”或“正在运行”。
📖 用法
访问气流UI
- 打开浏览器： http://localhost:8080
- 登录方式：
- 用户名: admin - 密码: admin123
DAGs示例
该项目包括两个示例DAGs：
1. example_hello_world
- 目的：用于测试基本功能的简单DAG
- 任务：巴什你好→ python你好→ print_context→ 再见
- 日程：仅手动触发
- 特性：演示基本的bash和Python运算符
如何测试:
- 导航到Airflow UI中的DAG页面
- 找到
example_hello_world - 单击播放按钮(▶️) 触发
- 在图形视图中查看日志
2. example_etl_pipeline
- 目的：完整的ETL工作流程演示
- 任务：开始→ 提取→ 变换→ 验证→ load → 通知→ end
- 日程：每日（或手动触发）
- 特性：显示任务间数据传递的XCom使用情况
如何测试:
- 转到DAGs页面
- 找到
example_etl_pipeline - 将开/关开关切换到“开”（用于自动调度）
- 点击播放按钮手动触发
- 在图形或网格视图中监视执行情况
查看日志
# All services
docker-compose logs -f
# Specific service
docker-compose logs -f airflow-webserver
docker-compose logs -f mcp-server
docker-compose logs -f airflow-scheduler停止服务
使用提供的脚本：
./stop.sh或手动：
# Keep data
docker-compose down
# Remove all data and volumes
docker-compose down -v🤖 MCP服务器配置
设置曲速终端
选项1：使用mcp-config.json（推荐）
- 查看MCP配置:
cat mcp-config.json- 添加到扭曲:
- 打开扭曲设置 - 导航到“功能”→ “MCP服务器” - 从添加配置 mcp-config.json
- 重新启动Warp终端
选项2：手动配置
将此配置添加到扭曲MCP设置中：
{
"mcpServers": {
"airflow": {
"command": "docker",
"args": [
"exec",
"-i",
"mcp-server",
"python",
"/app/server.py"
],
"env": {
"AIRFLOW_BASE_URL": "http://localhost:8080",
"AIRFLOW_API_USERNAME": "admin",
"AIRFLOW_API_PASSWORD": "admin123"
}
}
}
}测试MCP服务器
配置后，您可以在Warp中使用AI命令：
# List all DAGs
"Show me all Airflow DAGs"
# Trigger a DAG
"Trigger the example_hello_world DAG"
# Check DAG status
"What's the status of the latest run of example_etl_pipeline?"
# Get task logs
"Show me logs for the transform task in example_etl_pipeline"
# Manage variables
"Set Airflow variable API_KEY to test123"
"What's the value of API_KEY variable?"AI交互示例
场景1：使用配置触发DAG
AI: "Trigger example_hello_world with config message='Hello from AI'"场景2：监视DAG运行
AI: "Show me the last 5 runs of example_etl_pipeline and their status"场景3：调试失败的任务
AI: "The ETL pipeline failed, show me logs from the transform task"场景4：管理配置
AI: "List all Airflow variables and their values"场景5：上传新的DAG
AI: "Create a new DAG file called my_custom_dag.py with a simple hello world task"场景6：列出并读取DAG文件
AI: "Show me all DAG files and then read the content of example_hello_world.py"场景7：上传前验证DAG
AI: "Validate this DAG code before I upload it: [paste code]"场景8：暂停/取消暂停DAG
AI: "Pause the example_etl_pipeline DAG temporarily"📁 项目结构
airflow-deploy-mcp/
├── docker-compose.yaml # Docker Compose configuration
├── .env # Environment variables (gitignored)
├── .env.example # Environment template
├── .gitignore # Git ignore patterns
├── mcp-config.json # MCP Server configuration for Warp
├── README.md # This file
├── LICENSE # MIT License
├── start.sh # Startup script
├── stop.sh # Shutdown script
│
├── dags/ # Airflow DAGs directory
│ ├── example_hello_world.py
│ └── example_etl_pipeline.py
│
├── logs/ # Airflow logs (auto-generated)
├── plugins/ # Airflow plugins (optional)
├── config/ # Airflow configuration (optional)
│
└── mcp-server/ # MCP Server implementation
├── Dockerfile # MCP Server Docker image
├── requirements.txt # Python dependencies
└── server.py # MCP Server implementation🔧 故障排除
服务无法启动
检查Docker:
docker info检查内存分配:
- Docker 桌面版→ 设置→ 资源
- 推荐：内存>=4GB
检查初始化日志:
docker-compose logs airflow-init
docker-compose logs airflow-webserver气流用户界面不可访问
等待初始化完成:
docker-compose logs -f airflow-webserver寻找： "Running the Gunicorn Server"
检查端口冲突:
lsof -i :8080DAGs未出现在UI中
验证DAG文件:
ls -la dags/检查调度程序日志:
docker-compose logs -f airflow-scheduler刷新DAGs:
- 气流UI→ 管理员→ 全部刷新
MCP服务器问题
检查MCP容器状态:
docker-compose ps mcp-server
docker-compose logs mcp-server测试MCP服务器连接:
docker exec -it mcp-server python -c "import requests; print(requests.get('http://localhost:3000/health').status_code)"验证翘曲配置:
- 检查
mcp-config.json语法 - 重新启动Warp终端
- 查看扭曲日志
数据库问题
重置数据库 （使用脚本）：
./stop.sh
# Select "Yes" when prompted to remove volumes
./start.sh手动数据库重置:
docker-compose down -v
docker volume rm airflow-deploy-mcp_postgres-db-volume
./start.sh权限错误
修复文件夹权限:
mkdir -p logs dags plugins
chmod -R 755 logs dags plugins验证气流_UID:
grep AIRFLOW_UID .env
# Should match your user ID
echo $(id -u)🔒 安全说明
⚠️ 对生产部署很重要:
- 更改默认密码:
- 管理员密码 .env - PostgreSQL凭据 - Fernet密钥和秘密密钥
- 启用HTTPS:
- 使用反向代理（nginx、traefik） - 配置SSL证书
- 网络安全:
- 使用Docker网络隔离 - 配置防火墙规则 - 限制端口访问
- 秘密管理:
- 使用气流秘密后端 - 实施Docker机密或环境加密 - 切勿将敏感数据存储在DAG代码中
📚 文档
本项目
- 使用示例 -使用MCP Server进行DAG管理的实例
阿帕奇气流
MCP（模型上下文协议）
码头工人
🤝 贡献
欢迎投稿！请随时提交拉取请求。
📄 许可证
此项目根据MIT许可证获得许可-请参阅 许可证 文件以获取详细信息。
🆘 支持
如果您遇到问题或需要帮助：
- 检查 故障排除 章节
- 查看日志
docker-compose logs - 检查气流UI→ 浏览→ 任务实例日志
______________________________________________________________________
编排愉快！ 🎉