Astro Airflow

astro-airflow

Read-only debugging surface for Airflow on Astro. Pair three tools:

When NOT to use: local-only Airflow (astro dev start, parse, pytest) - that is vd:managing-astro-local-env. YAML authoring is vd:dag-factory. This skill is remote (staging/prod) inspection, plus wiring af at those URLs.

Prerequisites

• astro CLI ≥ 1.42, logged in (astro login; verify with astro context list)
• curl + jq
• af optional but preferred: uv tool install astro-airflow-mcp (one-shot: uvx --from astro-airflow-mcp af)
• A Deployment / Workspace / Organization API token in gopass for curl fallback. Mint via Astro UI → Deployment → Access → API Tokens. Least-privilege: deployment.get + deployment.airflow.*.get.

Always resolve the live deployment ID first. IDs in project docs go stale.

astro deployment list
astro deployment inspect <deployment-id> --key metadata.airflow_api_url

Choose the interface

1. Named Otto / upgrade / long investigation → vd:delegating-to-otto.
2. af on PATH (or uvx --from astro-airflow-mcp af) and a configured instance → use af. See references/af.md.
3. Platform / container logs / env vars → astro deployment ... below.
4. Otherwise → curl fallback.

af instance discover creates API tokens in Astro Cloud. Always --dry-run first and get explicit approval before a real discover.

uvx --from astro-airflow-mcp af instance list
uvx --from astro-airflow-mcp af instance discover --dry-run
# only after the user says yes:
# uvx --from astro-airflow-mcp af instance discover astro
uvx --from astro-airflow-mcp af instance use <name>

Add by hand without persisting the secret. af instance add --token writes the value into ~/.astro/config.yaml. Prefer a ${VAR} reference or a one-shot env:

export AIRFLOW_AUTH_TOKEN="$(gopass show -o <path/to/deployment-token>)"
export AIRFLOW_API_URL="https://<org>.astronomer.run/<short-id>"
uvx --from astro-airflow-mcp af dags list

Persistent instance (token stays in the environment, not the file):

# ~/.astro/config.yaml  (or project .astro/config.yaml)
instances:
- name: staging
  url: https://<org>.astronomer.run/<short-id>
  auth:
    token: ${AIRFLOW_AUTH_TOKEN}

If af prints reading from the legacy ~/.af/config.yaml, tell the user af migrate exists; do not run it unasked.

Platform ops: astro CLI

Container logs

Component is a boolean flag or --component <name> (CLI 1.45+). Airflow 3.3+ parse/import lives on --dag-processor, not only --scheduler. Airflow 3.x API is --apiserver; --webserver is Airflow 2.x.

--keyword is an exact phrase, not a regex. Do not combine --error/--warn/--info with each other or with --keyword - the CLI prints usage and exits. Official help examples that show --error --info are wrong on 1.45.x.

--error is not a reliable level filter: it can return [info] lines whose text contains "error" (DAG ids like el_twilio__error_code). Prefer --keyword "ImportError" on --dag-processor.

astro deployment logs <deployment-id> --scheduler --log-count 100
astro deployment logs <deployment-id> --dag-processor --keyword "ImportError"
astro deployment logs <deployment-id> --component scheduler --log-count 50   # CLI 1.45+
astro deployment logs <deployment-id> --apiserver --log-count 50
astro deployment logs <deployment-id> --triggerer --error
astro deployment logs <deployment-id> --workers --keyword "OOMKilled"

Use these when:

• DAGs not appearing / parse errors → --dag-processor --keyword "ImportError" (and --scheduler on older runtimes)
• Triggerer crashing → --triggerer --error
• Worker OOM → --workers --keyword "OOMKilled"

Environment variables and deploy state

astro deployment variable list --deployment-id <id>            # values redacted
astro deployment variable list --deployment-id <id> -s         # secrets (sensitive)
astro deployment inspect <id>
astro deployment pool list --deployment-id <id>

Create/update/copy variables only when the user explicitly asks. Confirm the deployment ID first; Astronomer has no undo.

DAG-level: prefer af

Once an instance points at the target deployment:

af health
af dags errors
af dags list
af runs list --dag-id <dag_id>
af runs diagnose <dag_id> <run_id>
af tasks logs <dag_id> <run_id> <task_id>
af tasks logs <dag_id> <run_id> <task_id> --try 2
af config pools
af api ls --filter xcom

Mutations (af dags unpause, af runs trigger, af runs clear, af runs delete) are opt-in - only when the user explicitly asks. Default is read-only.

Full command map: references/af.md.

Curl fallback (Airflow REST)

Use when af is missing or has no remote instance. Airflow 3.x = /api/v2/, Airflow 2.x = /api/v1/.

export ASTRO_TOKEN="$(gopass show -o <path/to/deployment-token>)"
export AF_URL="https://<org>.astronomer.run/<deployment-short-id>"
afcurl() { curl -fsSL -H "Authorization: Bearer ${ASTRO_TOKEN}" "${AF_URL}$1"; }

Do not name the wrapper af() - that shadows the real CLI.

Any Astro API token works as Authorization: Bearer (Deployment preferred, then Workspace, then Organization). The same value works for the astro CLI via ASTRO_API_TOKEN.

Runs and tasks

URL-encode run_id - scheduled IDs contain + / :.

afcurl "/api/v2/dags/~/dagRuns?limit=20&order_by=-start_date" \
  | jq '.dag_runs[] | {dag_id, run_id, state, start_date}'

afcurl "/api/v2/dags/<dag_id>/dagRuns?limit=10&order_by=-start_date" | jq
afcurl "/api/v2/dags/<dag_id>/dagRuns?state=failed&start_date_gte=2026-05-01T00:00:00Z" | jq
afcurl "/api/v2/dags/<dag_id>/dagRuns/${RUN_ID}/taskInstances?state=failed" | jq

Task logs (Airflow 3.x: content is events, not a string)

Verified on Airflow 3.1-3.3: the log endpoint returns {"content":[{event, timestamp, sources, ...}, ...], "continuation_token":"..."}. jq -r '.content' prints nothing useful. Iterate .content[] | .event.

afcurl "/api/v2/dags/<dag_id>/dagRuns/<run_id>/taskInstances/<task_id>/logs/<try_number>?full_content=true" \
  | jq -r '.content[] | select(type=="object") | .event' | grep -v '^::' | tail -n 200

full_content=true returns the first full block. Replay continuation_token as ?token= until it stops advancing. Running tasks do return logs mid-run.

Airflow 2.x /api/v1/ still returns .content as a plain string.

Other reads

afcurl "/api/v2/dags?limit=50&only_active=true"
afcurl "/api/v2/dags/<dag_id>/details"
afcurl "/api/v2/importErrors"
afcurl "/api/v2/connections"
afcurl "/api/v2/variables"     # values included - do not paste secrets
afcurl "/api/v2/pools"
afcurl "/api/v2/monitor/health"

Mutations (opt-in)

afw() { curl -fsSL -X "$1" -H "Authorization: Bearer ${ASTRO_TOKEN}" -H "Content-Type: application/json" "${AF_URL}$2" -d "$3"; }

afw PATCH "/api/v2/dags/<dag_id>?update_mask=is_paused" '{"is_paused": false}'
afw POST  "/api/v2/dags/<dag_id>/dagRuns" '{"dag_run_id":"manual__e2e","logical_date":null}'
afw PATCH "/api/v2/dags/<dag_id>/dagRuns/<run_id>" '{"state":"failed"}'

max_active_runs=1: unpausing can spawn a scheduled run, so a manual trigger sits queued behind it. Terminate the redundant queued run if the user wants only one.

Decision tree

User wants...                            → Use
─────────────────────────────────────────────────────────────────────
"use Otto" / AF2→3 upgrade / long audit  → vd:delegating-to-otto
"why did this run fail"                  → af runs diagnose  (else curl dagRuns → failed TIs → logs)
"any failed DAGs today"                  → af runs list / curl /dagRuns?state=failed
"scheduler broken / DAGs not parsing"    → astro logs --dag-processor  AND  af dags errors
"task log for try 2 of X"                → af tasks logs ... --try 2
"worker OOM"                             → astro logs --workers --keyword OOMKilled
"what env vars are set"                  → astro deployment variable list
"pool is starved"                        → af config pools
"trigger / clear failed"                 → only if user asks; prefer af, else curl. Never with a read-only token.

Investigate "DAG X failed"

# 1. most recent failed run
af runs list --dag-id <dag_id>
# fallback:
RUN_ID=$(afcurl "/api/v2/dags/<dag_id>/dagRuns?state=failed&limit=1&order_by=-start_date" \
         | jq -r '.dag_runs[0].run_id')

# 2. diagnose (af) or list failed tasks (curl)
af runs diagnose <dag_id> "$RUN_ID"
afcurl "/api/v2/dags/<dag_id>/dagRuns/${RUN_ID}/taskInstances?state=failed" \
  | jq '.task_instances[] | {task_id, try_number}'

# 3. logs - NEVER jq -r '.content' on Airflow 3
af tasks logs <dag_id> "$RUN_ID" <task_id>
afcurl "/api/v2/dags/<dag_id>/dagRuns/${RUN_ID}/taskInstances/<task_id>/logs/<try>?full_content=true" \
  | jq -r '.content[] | select(type=="object") | .event' | grep -v '^::' | tail -n 200

Safety rules

• Read-only by default. Do not trigger, clear, pause, or update variables unless the user asks.
• Token discipline. Pull from gopass. Never echo $ASTRO_TOKEN. Never write it outside the password store. Never commit it.
• Distinct tokens per environment.
• --keyword is an exact phrase. Do not pass foo|bar regex.
• Log fetch is heavy. Tail with | tail -n 200 unless asked for the full log.
• Do not paste secret variable values into chat.
• Confirm deployment ID (astro deployment list) before any mutation.

Failure modes

Discovery

astro version
astro context list
astro deployment list
astro deployment inspect <id> --key metadata.airflow_api_url
uvx --from astro-airflow-mcp af instance list

References

• af command map: references/af.md
• Official Airflow ops skill (af-centric): https://github.com/astronomer/agents/blob/main/skills/airflow/SKILL.md
• Astro CLI: https://docs.astronomer.io/astro/cli/overview
• Airflow 3 REST API: https://airflow.apache.org/docs/apache-airflow/stable/stable-rest-api-ref.html
• Otto delegation: vd:delegating-to-otto