需要查询、管理或排查 Apache Airflow（DAG…

通过 af CLI 对 Airflow 实例执行查询、触发、重试、日志读取、故障诊断、配置检查和 API 探索。

适用场景

解决思路

1. 使用 Astro CLI 初始化新项目：`astro dev init`
2. 使用 Astro CLI 启动本地 Airflow（webserver 位于 http://localhost:8080）：`astro dev start`
3. 使用 Astro CLI 解析 DAG 以快速捕获错误（无需启动 Airflow）：`astro dev parse`
4. 使用 Astro CLI 对 DAG 运行 pytest：`astro dev pytest`
5. 使用 Astro CLI 部署到生产：`astro deploy`（全量部署：镜像 + DAG）
6. 使用 Astro CLI 仅部署 DAG：`astro deploy --dags`（快速，无镜像构建）
7. 新项目参考 setting-up-astro-project 技能；本地环境参考 managing-astro-local-env 技能；部署参考 deploying-airflow 技能
8. 这些命令假设 `af` 在 PATH 上。通过 `astro otto` 自动获取，或使用 `uv tool install astro-airflow-mcp` 独立安装
9. 添加新实例：`af instance add prod --url https://airflow.example.com --token "$API_TOKEN"`
10. 添加带用户名密码的实例：`af instance add staging --url https://staging.example.com --username admin --password admin`
11. 列出所有实例：`af instance list`（以表格显示）
12. 切换实例：`af instance use prod`
13. 显示当前实例：`af instance current`
14. 删除实例：`af instance delete old-instance`
15. 自动发现实例前先用 `--dry-run` 预览：`af instance discover --dry-run` 预览所有可发现实例
16. 从所有后端发现实例：`af instance discover`（从 astro、local 发现）
17. 仅发现 Astro 部署：`af instance discover astro`
18. 包含所有可访问工作区：`af instance discover astro --all-workspaces`
19. 扫描常见本地 Airflow 端口：`af instance discover local`
20. 深度扫描所有端口 1024-65535：`af instance discover local --scan`
21. 重要：始终先运行 `--dry-run` 并征求用户同意后再运行不带 `--dry-run` 的 discover；非 dry-run 模式会在 Astro Cloud 创建 API token，这是敏感操作，需要明确批准
22. 显示实例来源（文件路径 + 作用域）：`af instance show prod`
23. 通过环境变量为单个命令覆盖实例：`AIRFLOW_API_URL=https://staging.example.com AIRFLOW_AUTH_TOKEN=$STG af dags list`
24. 或持久切换实例：`af instance use staging`
25. 配置布局镜像 git config system/global/local：Global 为 `~/.astro/config.yaml`（n/a，每用户）；Project shared 为 `<root>/.astro/config.yaml`（是，提交）；Project local 为 `<root>/.astro/config.local.yaml`（否，gitignored）
26. `<root>` 通过从 cwd 向上查找 `.astro/` 找到。项目内默认写入路由：`add`/`discover` → project-shared，`use` → project-local。用 `--global` / `--project` / `--local` 覆盖。设置 `AF_CONFIG=<path>` 绕过分层并使用单个文件
27. 从旧版 `~/.af/config.yaml` 迁移：使用 `af migrate`（幂等；将旧文件重命名为 `.bak`）
28. 配置中的 token 可以使用 `${VAR}` 语法引用环境变量：在配置文件中写 `instances: - name: prod url: https://airflow.example.com auth: token: ${AIRFLOW_API_TOKEN}`
29. 或直接使用环境变量（无需配置文件）：`export AIRFLOW_API_URL=http://localhost:8080`，`export AIRFLOW_AUTH_TOKEN=your-token-here`，或 `export AIRFLOW_USERNAME=admin`，`export AIRFLOW_PASSWORD=admin`
30. 或使用 CLI flags：`af --airflow-url http://localhost:8080 --token "$TOKEN" <command>`
31. `af health`：系统健康检查
32. `af dags list`：列出所有 DAG
33. `af dags get <dag_id>`：获取 DAG 详情
34. `af dags explore <dag_id>`：完整 DAG 调查
35. `af dags source <dag_id>`：获取 DAG 源代码
36. `af dags pause <dag_id>`：暂停 DAG 调度
37. `af dags unpause <dag_id>`：恢复 DAG 调度
38. `af dags errors`：列出导入错误
39. `af dags warnings`：列出 DAG 警告
40. `af dags stats`：DAG 运行统计
41. `af runs list`：列出 DAG 运行
42. `af runs get <dag_id> <run_id>`：获取运行详情
43. `af runs trigger <dag_id>`：触发 DAG 运行
44. `af runs trigger-wait <dag_id>`：触发并等待完成
45. `af runs delete <dag_id> <run_id>`：永久删除 DAG 运行
46. `af runs clear <dag_id> <run_id>`：清除运行以重新执行
47. `af runs diagnose <dag_id> <run_id>`：诊断失败的运行
48. `af tasks list <dag_id>`：列出 DAG 中的任务
49. `af tasks get <dag_id> <task_id>`：获取任务定义
50. `af tasks instance <dag_id> <run_id> <task_id>`：获取任务实例
51. `af tasks logs <dag_id> <run_id> <task_id>`：获取任务日志
52. `af config version`：Airflow 版本
53. `af config show`：完整配置
54. `af config connections`：列出连接
55. `af config variables`：列出变量
56. `af config variable <key>`：获取特定变量
57. `af config pools`：列出池
58. `af config pool <name>`：获取池详情
59. `af config plugins`：列出插件
60. `af config providers`：列出提供者
61. `af config assets`：列出 assets/datasets
62. `af api <endpoint>`：直接 REST API 访问
63. `af api ls`：列出可用 API 端点
64. `af api ls --filter X`：列出匹配模式的端点
65. `af registry providers`：列出 Airflow Registry 中的提供者
66. `af registry modules <provider>`：列出提供者中的 operators/hooks/sensors/transfers
67. `af registry parameters <provider>`：提供者类的构造函数签名（name, type, default, required）
68. `af registry connections <provider>`：提供者暴露的连接类型
69. 意图“如何本地运行 Airflow？”/“设置 Airflow” -> 使用 managing-astro-local-env 技能（使用 Astro CLI）
70. 意图“创建新 Airflow 项目”/“初始化项目” -> 使用 setting-up-astro-project 技能（使用 Astro CLI）
71. 意图“如何安装 Airflow？”/“开始使用 Airflow” -> 使用 setting-up-astro-project 技能
72. 意图“存在哪些 DAG？”/“列出所有 DAG” -> `af dags list`
73. 意图“告诉我关于 DAG X”/“DAG Y 是什么？” -> `af dags explore <dag_id>`
74. 意图“DAG X 的调度是什么？” -> `af dags get <dag_id>`
75. 意图“显示 DAG X 的代码” -> `af dags source <dag_id>`
76. 意图“停止 DAG X”/“暂停此工作流” -> `af dags pause <dag_id>`
77. 意图“恢复 DAG X” -> `af dags unpause <dag_id>`
78. 意图“有 DAG 错误吗？” -> `af dags errors`
79. 意图“创建新 DAG”/“编写管道” -> 使用 authoring-dags 技能
80. 意图“执行了哪些运行？” -> `af runs list`
81. 意图“运行 DAG X”/“触发管道” -> `af runs trigger <dag_id>`
82. 意图“运行 DAG X 并等待” -> `af runs trigger-wait <dag_id>`
83. 意图“为什么这次运行失败？” -> `af runs diagnose <dag_id> <run_id>`
84. 意图“删除这次运行”/“移除卡住的运行” -> `af runs delete <dag_id> <run_id>`
85. 意图“清除这次运行”/“重试这次运行”/“重新运行” -> `af runs clear <dag_id> <run_id>`
86. 意图“测试此 DAG 并在失败时修复” -> 使用 testing-dags 技能
87. 意图“DAG X 中有哪些任务？” -> `af tasks list <dag_id>`
88. 意图“获取任务日志”/“为什么任务失败？” -> `af tasks logs <dag_id> <run_id> <task_id>`
89. 意图“完整根因分析”/“诊断并修复” -> 使用 debugging-dags 技能
90. 意图“数据是否新鲜？”/“这张表上次更新是什么时候？” -> 使用 checking-freshness 技能
91. 意图“这些数据来自哪里？” -> 使用 tracing-upstream-lineage 技能
92. 意图“什么依赖这张表？”/“如果我改变这个会破坏什么？” -> 使用 tracing-downstream-lineage 技能
93. 意图“部署我的 DAG”/“推送到生产” -> 使用 deploying-airflow 技能
94. 意图“设置 CI/CD”/“自动化部署” -> 使用 deploying-airflow 技能
95. 意图“部署到 Kubernetes”/“设置 Helm” -> 使用 deploying-airflow 技能
96. 意图“astro deploy”/“仅 DAG 部署” -> 使用 deploying-airflow 技能
97. 意图“Airflow 是什么版本？” -> `af config version`
98. 意图“存在哪些连接？” -> `af config connections`
99. 意图“池是否满了？” -> `af config pools`
100. 意图“Airflow 是否健康？” -> `af health`
101. 意图“有哪些 API 端点可用？” -> `af api ls`
102. 意图“查找变量端点” -> `af api ls --filter variable`
103. 意图“访问 XCom 值”/“获取 XCom” -> `af api xcom-entries -F dag_id=X -F task_id=Y`
104. 意图“获取事件日志”/“审计追踪” -> `af api event-logs -F dag_id=X`
105. 意图“通过 API 创建连接” -> `af api connections -X POST --body '{...}'`
106. 意图“通过 API 创建变量” -> `af api variables -X POST -F key=name -f value=val`
107. 意图“提供者 X 有哪些 operators？” -> `af registry modules <provider>`
108. 意图“operator Y 的构造函数参数是什么？” -> `af registry parameters <provider>`
109. 意图“存在哪些提供者？”/“是否有 Z 的提供者？” -> `af registry providers`
110. 意图“提供者 X 暴露哪些连接类型？” -> `af registry connections <provider>`
111. 意图“编写使用特定 operator 的 DAG” -> 使用 registry 验证当前签名，然后再复制示例
112. 如果使用 Astro CLI，无需运行中的 Airflow 实例即可验证 DAG：`astro dev parse` 捕获导入错误和语法问题
113. 如果使用 Astro CLI，运行单元测试：`astro dev pytest`
114. 否则，针对运行中的实例验证：`af dags errors` 检查解析/导入错误
115. 否则，检查弃用警告：`af dags warnings`
116. Airflow Registry 位于 `airflow.apache.org/registry`，是提供者类及其当前构造函数签名的权威来源；编写 DAG 时优先使用它而非记忆或过时文档，因为 registry 反映实时提供者发布
117. 列出所有提供者并选择所需：`af registry providers | jq '.providers[] | {id, name, version}'`
118. 列出提供者中每个 operator/hook/sensor（例如 standard、amazon、google）：`af registry modules standard | jq '.modules[] | {name, type, import_path, docs_url}'`
119. 获取特定类的当前构造函数签名：`af registry parameters standard | jq '.classes["airflow.providers.standard.operators.hitl.ApprovalOperator"].parameters'`
120. 按子字符串过滤模块（当知道概念但不知道类名时有用）：`af registry modules standard | jq '.modules[] | select(.import_path | test("hitl"))'`
121. 结果本地缓存：最新版本 1 小时，固定版本 30 天（不可变）。在任何 `modules` / `parameters` / `connections` 调用中添加 `--version X.Y.Z` 以指定特定发布版本
122. 列出最近运行以找到失败：`af runs list --dag-id my_dag`
123. 诊断特定运行：`af runs diagnose my_dag manual__2024-01-15T10:00:00+00:00`
124. 获取失败任务的日志（来自诊断输出）：`af tasks logs my_dag manual__2024-01-15T10:00:00+00:00 extract_data`
125. 修复后，清除运行以重试所有任务：`af runs clear my_dag manual__2024-01-15T10:00:00+00:00`
126. 整体系统健康：`af health`
127. 检查损坏的 DAG：`af dags errors`
128. 检查池利用率：`af config pools`
129. 获取全面概览（元数据 + 任务 + 源代码）：`af dags explore my_dag`
130. 检查是否暂停：`af dags get my_dag`
131. 检查导入错误：`af dags errors`
132. 检查最近运行：`af runs list --dag-id my_dag`
133. 选项 1：触发并等待（阻塞）：`af runs trigger-wait my_dag --timeout 1800`
134. 选项 2：触发后稍后检查：`af runs trigger my_dag`，然后 `af runs get my_dag <run_id>`
135. 所有命令输出 JSON（除了 `instance` 命令使用人类可读表格）：例如 `af dags list` 输出 `{ "total_dags": 5, "returned_count": 5, "dags": [...] }`
136. 使用 `jq` 过滤：查找失败的运行：`af runs list | jq '.dag_runs[] | select(.state == "failed")'`
137. 使用 `jq` 仅获取 DAG ID：`af dags list | jq '.dags[].dag_id'`
138. 使用 `jq` 查找暂停的 DAG：`af dags list | jq '[.dags[] | select(.is_paused == true)]'`
139. 获取特定重试尝试的日志：`af tasks logs my_dag run_id task_id --try 2`
140. 获取映射任务索引的日志：`af tasks logs my_dag run_id task_id --map-index 5`
141. 对高级命令未覆盖的端点（XCom、event-logs、backfills 等）使用 `af api`
142. 发现可用端点：`af api ls`
143. 按模式发现端点：`af api ls --filter variable`
144. 基本用法：`af api dags`；`af api dags -F limit=10 -F only_active=true`；`af api variables -X POST -F key=my_var -f value="my value"`；`af api variables/old_var -X DELETE`
145. 字段语法：`-F key=value` 自动转换类型，`-f key=value` 保持为字符串
146. 完整参考：见 `api-reference.md` 获取所有选项、常见端点（XCom、event-logs、backfills）和示例
147. 相关技能路由：authoring-dags 用于创建或编辑 DAG 文件；testing-dags 用于迭代测试-调试-修复-重测；debugging-dags 用于深度根因分析和故障诊断；checking-freshness 用于检查数据是否最新或过期；tracing-upstream-lineage 用于查找数据来源；tracing-downstream-lineage 用于影响分析；deploying-airflow 用于部署 DAG 到生产；migrating-airflow-2-to-3 用于升级 DAG；managing-astro-local-env 用于启动、停止或排查本地 Airflow；setting-up-astro-project 用于初始化新 Astro/Airflow 项目；airflow-state-store 用于每任务检查点、水印、崩溃安全 operator（Airflow 3.3+）；airflow-hitl 用于暂停 DAG 等待人工批准或输入（Airflow 3.1+）

适用边界 · 注意事项

• 不要用于对 Airflow 元数据表做仓库/SQL 分析（应使用 analyzing-data）。
• 不要用于深度根因报告（应使用 debugging-dags 或 airflow-investigation）。
• 不要用于 DAG 编写、测试、部署或 Airflow 2 到 3 迁移（应使用 authoring-dags、testing-dags、deploying-airflow、migrating-airflow-2-to-3 等对应技能）。