npx skills add ...
npx skills add astronomer/agents --skill airflow
npx skills add astronomer/agents --skill airflow
Queries, manages, and troubleshoots Apache Airflow using the `af` CLI. Use when working with anything related to Airflow - a DAG, a DAG run, a task log, an import or parse error, a broken DAG, or any Airflow operation. Covers listing and triggering DAGs, retrying runs, reading task logs, diagnosing failures, debugging import and parse errors, checking connections, variables and pools, exploring the REST API, and monitoring health (for example "trigger a pipeline", "retry a run", "list connections", "check Airflow health", "why did my DAG fail"). This is the entrypoint that routes to sibling skills for authoring, testing, deploying, and migrating Airflow 2 to 3. Not for warehouse/SQL analytics on Airflow metadata tables (use analyzing-data); for deep root-cause reports use debugging-dags or airflow-investigation.
Use af commands to query, manage, and troubleshoot Airflow workflows.
The Astro CLI is the recommended way to run Airflow locally and deploy to production. It provides a containerized Airflow environment that works out of the box:
For more details:
These commands assume af is on PATH. Run via astro otto to get it automatically, or install standalone with uv tool install astro-airflow-mcp.
Manage multiple Airflow instances with persistent configuration:
Config layout (mirrors git config system/global/local):
| Scope | File | Committed? |
|---|---|---|
| Global | ~/.astro/config.yaml | n/a (per-user) |
| Project shared | <root>/.astro/config.yaml | yes |
| Project local | <root>/.astro/config.local.yaml | no (gitignored) |
<root> is found by walking up from cwd looking for .astro/. Default write routing inside a project: add/discover → project-shared, use → project-local. Override with --global / --project / --local. Set AF_CONFIG=<path> to bypass layering and use a single file.
Migrate from the legacy ~/.af/config.yaml with af migrate (idempotent; renames the old file to .bak).
Tokens in config can reference environment variables using ${VAR} syntax:
Or use environment variables directly (no config file needed):
Or CLI flags: af --airflow-url http://localhost:8080 --token "$TOKEN" <command>
| Command | Description |
|---|---|
af health | System health check |
af dags list | List all DAGs |
af dags get <dag_id> | Get DAG details |
af dags explore <dag_id> | Full DAG investigation |
af dags source <dag_id> | Get DAG source code |
af dags pause <dag_id> | Pause DAG scheduling |
af dags unpause <dag_id> | Resume DAG scheduling |
af dags errors | List import errors |
af dags warnings | List DAG warnings |
af dags stats | DAG run statistics |
af runs list | List DAG runs |
af runs get <dag_id> <run_id> | Get run details |
af runs trigger <dag_id> | Trigger a DAG run |
af runs trigger-wait <dag_id> | Trigger and wait for completion |
af runs delete <dag_id> <run_id> | Permanently delete a DAG run |
af runs clear <dag_id> <run_id> | Clear a run for re-execution |
af runs diagnose <dag_id> <run_id> | Diagnose failed run |
af tasks list <dag_id> | List tasks in DAG |
af tasks get <dag_id> <task_id> | Get task definition |
af tasks instance <dag_id> <run_id> <task_id> | Get task instance |
af tasks logs <dag_id> <run_id> <task_id> | Get task logs |
af config version | Airflow version |
af config show | Full configuration |
af config connections | List connections |
af config variables | List variables |
af config variable <key> | Get specific variable |
af config pools | List pools |
af config pool <name> | Get pool details |
af config plugins | List plugins |
af config providers | List providers |
af config assets | List assets/datasets |
af api <endpoint> | Direct REST API access |
af api ls | List available API endpoints |
af api ls --filter X | List endpoints matching pattern |
af registry providers | List providers in the Airflow Registry |
af registry modules <provider> | List operators/hooks/sensors/transfers in a provider |
af registry parameters <provider> | Constructor signatures (name, type, default, required) for a provider's classes |
af registry connections <provider> | Connection types a provider exposes |
af dags listaf dags explore <dag_id>af dags get <dag_id>af dags source <dag_id>af dags pause <dag_id>af dags unpause <dag_id>af dags errorsaf runs listaf runs trigger <dag_id>af runs trigger-wait <dag_id>af runs diagnose <dag_id> <run_id>af runs delete <dag_id> <run_id>af runs clear <dag_id> <run_id>af tasks list <dag_id>af tasks logs <dag_id> <run_id> <task_id>af config versionaf config connectionsaf config poolsaf healthaf api lsaf api ls --filter variableaf api xcom-entries -F dag_id=X -F task_id=Yaf api event-logs -F dag_id=Xaf api connections -X POST --body '{...}'af api variables -X POST -F key=name -f value=valaf registry modules <provider>af registry parameters <provider>af registry providersaf registry connections <provider>If you're using the Astro CLI, you can validate DAGs without a running Airflow instance:
Otherwise, validate against a running instance:
The Airflow Registry at airflow.apache.org/registry is the authoritative source for provider classes and their current constructor signatures. Prefer it over memory or stale documentation when authoring DAGs — the registry reflects the live provider release.
Results are cached locally: 1 hour for the latest version, 30 days for pinned versions (which are immutable). Add --version X.Y.Z to any modules / parameters / connections call to target a specific release.
All commands output JSON (except instance commands which use human-readable tables):
Use jq for filtering:
af apiUse af api for endpoints not covered by high-level commands (XCom, event-logs, backfills, etc).
Field syntax: -F key=value auto-converts types, -f key=value keeps as string.
Full reference: See api-reference.md for all options, common endpoints (XCom, event-logs, backfills), and examples.
| Skill | Use when... |
|---|---|
| authoring-dags | Creating or editing DAG files with best practices |
| testing-dags | Iterative test -> debug -> fix -> retest cycles |
| debugging-dags | Deep root cause analysis and failure diagnosis |
| checking-freshness | Checking if data is up to date or stale |
| tracing-upstream-lineage | Finding where data comes from |
| tracing-downstream-lineage | Impact analysis -- what breaks if something changes |
| deploying-airflow | Deploying DAGs to production (Astro, Docker Compose, Kubernetes) |
| migrating-airflow-2-to-3 | Upgrading DAGs from Airflow 2.x to 3.x |
| managing-astro-local-env | Starting, stopping, or troubleshooting local Airflow |
| setting-up-astro-project | Initializing a new Astro/Airflow project |
| airflow-state-store | Per-task checkpointing, watermarks, crash-safe operators (Airflow 3.3+) |
| airflow-hitl | Pausing a DAG for human approval or input (Airflow 3.1+) |
# Add a new instance
af instance add prod --url https://airflow.example.com --token "$API_TOKEN"
af instance add staging --url https://staging.example.com --username admin --password admin
# List and switch instances
af instance list # Shows all instances in a table
af instance use prod # Switch to prod instance
af instance current # Show current instance
af instance delete old-instance
# Auto-discover instances (use --dry-run to preview first)
af instance discover --dry-run # Preview all discoverable instances
af instance discover # Discover from all backends (astro, local)
af instance discover astro # Discover Astro deployments only
af instance discover astro --all-workspaces # Include all accessible workspaces
af instance discover local # Scan common local Airflow ports
af instance discover local --scan # Deep scan all ports 1024-65535
# IMPORTANT: Always run with --dry-run first and ask for user consent before
# running discover without it. The non-dry-run mode creates API tokens in
# Astro Cloud, which is a sensitive action that requires explicit approval.
# Show where an instance came from (file path + scope)
af instance show prod
# Override instance for a single command via env vars
AIRFLOW_API_URL=https://staging.example.com AIRFLOW_AUTH_TOKEN=$STG af dags list
# Or switch persistently
af instance use staginginstances:
- name: prod
url: https://airflow.example.com
auth:
token: ${AIRFLOW_API_TOKEN}export AIRFLOW_API_URL=http://localhost:8080
export AIRFLOW_AUTH_TOKEN=your-token-here
# Or username/password:
export AIRFLOW_USERNAME=admin
export AIRFLOW_PASSWORD=admin# Parse DAGs to catch import errors and syntax issues
astro dev parse
# Run unit tests
astro dev pytestaf dags errors # Check for parse/import errors
af dags warnings # Check for deprecation warnings# List all providers and pick the one you need
af registry providers | jq '.providers[] | {id, name, version}'
# List every operator / hook / sensor in a provider (e.g. standard, amazon, google)
af registry modules standard \
| jq '.modules[] | {name, type, import_path, docs_url}'
# Get the current constructor signature for a specific class
af registry parameters standard \
| jq '.classes["airflow.providers.standard.operators.hitl.ApprovalOperator"].parameters'
# Filter modules by substring (useful when you know the concept but not the class)
af registry modules standard \
| jq '.modules[] | select(.import_path | test("hitl"))'# 1. List recent runs to find failure
af runs list --dag-id my_dag
# 2. Diagnose the specific run
af runs diagnose my_dag manual__2024-01-15T10:00:00+00:00
# 3. Get logs for failed task (from diagnose output)
af tasks logs my_dag manual__2024-01-15T10:00:00+00:00 extract_data
# 4. After fixing, clear the run to retry all tasks
af runs clear my_dag manual__2024-01-15T10:00:00+00:00# 1. Overall system health
af health
# 2. Check for broken DAGs
af dags errors
# 3. Check pool utilization
af config pools# Get comprehensive overview (metadata + tasks + source)
af dags explore my_dag# Check if paused
af dags get my_dag
# Check for import errors
af dags errors
# Check recent runs
af runs list --dag-id my_dag# Option 1: Trigger and wait (blocking)
af runs trigger-wait my_dag --timeout 1800
# Option 2: Trigger and check later
af runs trigger my_dag
af runs get my_dag <run_id>af dags list
# {
# "total_dags": 5,
# "returned_count": 5,
# "dags": [...]
# }# Find failed runs
af runs list | jq '.dag_runs[] | select(.state == "failed")'
# Get DAG IDs only
af dags list | jq '.dags[].dag_id'
# Find paused DAGs
af dags list | jq '[.dags[] | select(.is_paused == true)]'# Get logs for specific retry attempt
af tasks logs my_dag run_id task_id --try 2
# Get logs for mapped task index
af tasks logs my_dag run_id task_id --map-index 5# Discover available endpoints
af api ls
af api ls --filter variable
# Basic usage
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