用途：面向数据工程师，在 Airflow DAG 运行失败时按结构化流程定位根因并给出可执行修复方案，同时提供预防建议。
主要能力： 一、定位失败。指定 DAG 时可运行 af runs diagnose <dag_id> <dag_run_id>；未给 run_id 用 af dags stats 查找近期失败；未指定 DAG 用 af health 查看全局失败，并用 af dags errors 检查导入错误。 二、获取错误详情。用 af tasks logs <dag_id> <dag_run_id> <task_id> 取任务日志，跳过 Airflow 样板找到真实异常，并把失败归类为数据、代码、基础设施或依赖问题。 三、排查上下文。检查近期代码部署与 git 历史、数据量变化、上游健康度、历史是否复发、失败时间点；用 af runs get <dag_id> <dag_run_id> 对比失败的与近期成功的运行。 四、依赖漂移排查。优先对比相邻 Astro 镜像的 pip freeze 与 python --version；venv 类算子（@task.virtualenv、PythonVirtualenvOperator、ExternalPythonOperator、KubernetesPodOperator）需直接查看其 requirements、image、python_version 参数，常见诱因是无上界约束、latest 标签或 Python 小版本变化；必要时按配置的索引（UV_INDEX_URL、PIP_INDEX_URL、pyproject.toml、pip.conf、Dockerfile）查询包发布时间，判断是否落在最后一次成功与首次失败之间。 五、输出结构。给出根因、影响评估（受影响数据、下游阻塞、是否影响生产报表）、即时修复步骤（SQL、代码改动或需联系/重启的对象）与预防措施（数据质量检查、错误处理、告警、文档、锁定依赖）。 六、常用命令。af runs clear <dag_id> <run_id> 重跑整个 DAG；af tasks clear <dag_id> <run_id> <task_ids> -D 重跑指定失败任务；af runs delete <dag_id> <run_id> 删除卡住或不需要的运行。
使用方式：命令假设 af 已在 PATH 中；可通过 astro otto 自动获得，或用 uv tool install astro-airflow-mcp 独立安装。在 Astro 上还可查看部署活动日志、配置 Astro 告警与可观测性面板；OSS Airflow 可用 DAGs 页面、Graph 视图与任务日志排查。
限制：资料未提供具体收费、适用版本范围与效果保证；实际命令可用性取决于环境配置，私有无索引可能无法用 PyPI 方式查询。