Imported from minyeamer/linkmerce (.agents/skills/airflow/SKILL.md). Install upstream with npx skills add minyeamer/linkmerce --skill airflow. Copyright stays with the author.

Airflow Operations

Use af commands to query, manage, and troubleshoot Airflow workflows.

Terminology

• Use Dag for an Airflow workflow in prose and Dags for the plural.
• Use DAG only for the Python class itself, including its import and constructor usage.
• Preserve code identifiers such as dag, dag_id, filenames, CLI arguments, and class names such as DagBag, DagModel, and DagRunApi.

Astro CLI

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:

# Initialize a new project
astro dev init

# Start local Airflow (webserver at http://localhost:8080)
astro dev start

# Parse Dags to catch errors quickly (no need to start Airflow)
astro dev parse

# Run pytest against your Dags
astro dev pytest

# Deploy to production
astro deploy            # Full deploy (image + Dags)
astro deploy --dags     # Dag-only deploy (fast, no image build)

For more details:

• New project? See the setting-up-astro-project skill
• Local environment? See the managing-astro-local-env skill
• Deploying? See the deploying-airflow skill

Running the CLI

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.

Instance Configuration

Manage multiple Airflow instances with persistent configuration:

# 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 staging

Config layout (mirrors git config system/global/local):

<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:

instances:
- name: prod
  url: https://airflow.example.com
  auth:
    token: ${AIRFLOW_API_TOKEN}

Or use environment variables directly (no config file needed):

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

Or CLI flags: af --airflow-url http://localhost:8080 --token "$TOKEN" <command>

Quick Reference

User Intent Patterns

Getting Started

• "How do I run Airflow locally?" / "Set up Airflow" -> use the managing-astro-local-env skill (uses Astro CLI)
• "Create a new Airflow project" / "Initialize project" -> use the setting-up-astro-project skill (uses Astro CLI)
• "How do I install Airflow?" / "Get started with Airflow" -> use the setting-up-astro-project skill

Dag Operations

• "What Dags exist?" / "List all Dags" -> af dags list
• "Tell me about Dag X" / "What is Dag Y?" -> af dags explore <dag_id>
• "What's the schedule for Dag X?" -> af dags get <dag_id>
• "Show me the code for Dag X" -> af dags source <dag_id>
• "Stop Dag X" / "Pause this workflow" -> af dags pause <dag_id>
• "Resume Dag X" -> af dags unpause <dag_id>
• "Are there any Dag errors?" -> af dags errors
• "Create a new Dag" / "Write a pipeline" -> use the authoring-dags skill

Run Operations

• "What runs have executed?" -> af runs list
• "Run Dag X" / "Trigger the pipeline" -> af runs trigger <dag_id>
• "Run Dag X and wait" -> af runs trigger-wait <dag_id>
• "Why did this run fail?" -> af runs diagnose <dag_id> <run_id>
• "Delete this run" / "Remove stuck run" -> af runs delete <dag_id> <run_id>
• "Clear this run" / "Retry this run" / "Re-run this" -> af runs clear <dag_id> <run_id>
• "Test this Dag and fix if it fails" -> use the testing-dags skill

Task Operations

• "What tasks are in Dag X?" -> af tasks list <dag_id>
• "Get task logs" / "Why did task fail?" -> af tasks logs <dag_id> <run_id> <task_id>
• "Full root cause analysis" / "Diagnose and fix" -> use the debugging-dags skill

Data Operations

• "Is the data fresh?" / "When was this table last updated?" -> use the checking-freshness skill
• "Where does this data come from?" -> use the tracing-upstream-lineage skill
• "What depends on this table?" / "What breaks if I change this?" -> use the tracing-downstream-lineage skill

Deployment Operations

• "Deploy my Dags" / "Push to production" -> use the deploying-airflow skill
• "Set up CI/CD" / "Automate deploys" -> use the deploying-airflow skill
• "Deploy to Kubernetes" / "Set up Helm" -> use the deploying-airflow skill
• "astro deploy" / "Dag-only deploy" -> use the deploying-airflow skill

System Operations

• "What version of Airflow?" -> af config version
• "What connections exist?" -> af config connections
• "Are pools full?" -> af config pools
• "Is Airflow healthy?" -> af health

API Exploration

• "What API endpoints are available?" -> af api ls
• "Find variable endpoints" -> af api ls --filter variable
• "Access XCom values" / "Get XCom" -> af api xcom-entries -F dag_id=X -F task_id=Y
• "Get event logs" / "Audit trail" -> af api event-logs -F dag_id=X
• "Create connection via API" -> af api connections -X POST --body '{...}'
• "Create variable via API" -> af api variables -X POST -F key=name -f value=val

Registry Discovery

• "What operators does provider X have?" -> af registry modules <provider>
• "What are the constructor params for operator Y?" -> af registry parameters <provider>
• "What providers exist?" / "Is there a provider for Z?" -> af registry providers
• "What connection types does provider X expose?" -> af registry connections <provider>
• "Writing a Dag with a specific operator" -> use registry to verify current signature before copying examples

Common Workflows

Validate Dags Before Deploying

If you're using the Astro CLI, you can validate Dags without a running Airflow instance:

# Parse Dags to catch import errors and syntax issues
astro dev parse

# Run unit tests
astro dev pytest

Otherwise, validate against a running instance:

af dags errors     # Check for parse/import errors
af dags warnings   # Check for deprecation warnings

Discover Operator Signatures Before Writing Code

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.

# 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"))'

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.

Investigate a Failed Run

# 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

Morning Health Check

# 1. Overall system health
af health

# 2. Check for broken Dags
af dags errors

# 3. Check pool utilization
af config pools

Understand a Dag

# Get comprehensive overview (metadata + tasks + source)
af dags explore my_dag

Check Why Dag Isn't Running

# 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

Trigger and Monitor

# 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>

Output Format

All commands output JSON (except instance commands which use human-readable tables):

af dags list
# {
#   "total_dags": 5,
#   "returned_count": 5,
#   "dags": [...]
# }

Use jq for filtering:

# 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)]'

Task Logs Options

# 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

Direct API Access with af api

Use af api for endpoints not covered by high-level commands (XCom, event-logs, backfills, etc).

# 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

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.

Related Skills