Airflow Tools

Overview

Tool Name

airflow_tools

Purpose

The airflow_tools group lets Genesis Data Agents inspect DAGs, review runs and task failures, fetch task logs or DAG source, pause or unpause DAGs, and trigger new runs through the Airflow REST API. Genesis exposes Airflow as first-class tools such as airflow_config, airflow_list_dags, and airflow_trigger rather than a single catch-all action tool. Airflow lineage scans also add DAG schedule, source-file, repository execution, and owner-principal context to the Digital Twin and Context Graph.

Connection Requirements

Create an Airflow, Astronomer, or Amazon MWAA connection in Config -> Connections under Data Pipelines before using these tools. All three connection types use the same airflow_tools group.

Classic Airflow (username/password)

Field	Stored key	Type	Required	Notes
`Airflow URL`	`host`	`string`	Yes	Example: `https://airflow.yourcompany.com`
`Username`	`username`	`string`	Yes	Basic-auth username
`Password`	`password`	`string`	Yes	Saved as a secret
`Airflow Version`	`version`	`string`	Yes	`2` or `3`
`Airflow Home`	`home`	`string`	No	Used by `airflow_dag_source` local fallback
`DAGs Path`	`dags_path`	`string`	No	Overrides `home/dags` for local DAG lookup

Astronomer Cloud (API token)

Field	Stored key	Type	Required	Notes
`Deployment URL`	`host`	`string`	Yes	Copy from Open Airflow in Astronomer Cloud; example: `https://<deployment-id>.<org>.astronomer.run/<slug>`
`API Token`	`api_token`	`string`	Yes	Saved as a secret; sent as a bearer token
`Airflow Version`	`version`	`string`	No	`2` or `3`; defaults to `3`

Amazon MWAA (IAM)

Genesis calls the environment’s Airflow REST API through the MWAA InvokeRestApi API with IAM credentials, so a private webserver works without network access to its VPC. Choose the Assume Role, Access Key, IAM Role (IRSA / Pod Identity), or Snowflake Workload Identity (keyless) mode. The first three take the AWS credential fields of that mode plus:

Field	Stored key	Type	Required	Notes
`Environment Name`	`environment_name`	`string`	Yes	As shown in the Amazon MWAA console
`Region`	`region`	`string`	Yes	AWS region of the environment

The Airflow version is read from the environment, so there is no version field. The IAM policy the role needs is on the Connections page. On a Snowflake Native App, Snowflake Workload Identity (keyless) uses a Snowflake workload identity secret instead of a stored AWS key. Provide the role ARN and fully qualified secret name, with an optional role session name. See the MWAA connection setup for the secret grants and IAM trust policy.

Set a user or role default on one Airflow, Astronomer, or Amazon MWAA connection if you want the tools to use it automatically when connection_id is omitted. For tools that expose connection_id, pass the saved connection name to override the default. Use airflow.<name>, astronomer.<name>, or mwaa.<name> when you want to be explicit about the connection type.

Airflow tools read credentials only from saved Airflow, Astronomer, or Amazon MWAA connections. They do not fall back to legacy shared entries, local ~/.genesis/airflow_config.json, or AIRFLOW_* environment variables.

airflow_config(action="GET") and airflow_config(action="TEST") use the requester-aware scoped default Airflow, Astronomer, or Amazon MWAA connection. If more than one of those types has a scoped default for the caller, specify connection_id on other tools (for example astronomer.production) or clear the extra defaults before using the config tools. Inline base_url, username, and password values are no longer accepted.

Function Reference

Genesis injects bot_id, user_id, and thread_id automatically. The tables below list only the user-supplied parameters.

Connection and Discovery

Tool	Purpose	Parameters
`airflow_config`	View the saved Airflow configuration or test connectivity.	`action` (`string`, required; `GET` or `TEST`)
`airflow_list_dags`	List DAGs visible to the selected Airflow connection.	`only_active` (`boolean`, default `true`), `tags` (`string[] \| null`, default `null`), `connection_id` (`string \| null`, default `null`)
`airflow_get_dag`	Fetch one DAG definition and include a UI link when available.	`dag_id` (`string`, required), `connection_id` (`string \| null`, default `null`)
`airflow_get_running_dags`	List currently running DAGs.	No user parameters
`airflow_status`	Return DAG status summaries, recent runs, or failed tasks for a DAG.	`dag_id` (`string \| null`, default `null`), `state` (`string \| null`, default `null`; `running`, `success`, or `failed`), `limit` (`integer`, default `20`)

Run and Failure Inspection

Tool	Purpose	Parameters
`airflow_get_dag_runs`	List runs for one DAG with optional state and date filtering.	`dag_id` (`string`, required), `limit` (`integer`, default `10`), `state` (`string \| null`, default `null`), `order_by` (`string \| null`, default `null`), `start_date` (`string \| null`, default `null`), `end_date` (`string \| null`, default `null`), `connection_id` (`string \| null`, default `null`)
`airflow_get_all_dag_runs`	List runs across all DAGs in a single request.	`limit` (`integer`, default `100`), `state` (`string \| null`, default `null`), `order_by` (`string \| null`, default `null`), `start_date` (`string \| null`, default `null`), `end_date` (`string \| null`, default `null`), `connection_id` (`string \| null`, default `null`)
`airflow_get_latest_dag_run`	Fetch the most recent run for a DAG.	`dag_id` (`string`, required), `connection_id` (`string \| null`, default `null`)
`airflow_get_failed_tasks`	Return failed tasks within a lookback window.	`dag_id` (`string \| null`, default `null`), `hours` (`integer`, default `24`)
`airflow_failure_details`	Collect failed tasks for a run and optionally include logs.	`dag_id` (`string`, required), `airflow_run_id` (`string \| null`, default `null`), `include_logs` (`boolean`, default `true`), `max_log_bytes` (`integer`, default `20000`)

Task Inspection

Tool	Purpose	Parameters
`airflow_get_task_instances`	List task instances for a DAG run.	`dag_id` (`string`, required), `airflow_run_id` (`string`, required), `state` (`string \| null`, default `null`)
`airflow_get_task_instance`	Fetch one task instance.	`dag_id` (`string`, required), `airflow_run_id` (`string`, required), `task_id` (`string`, required)
`airflow_get_task_logs`	Fetch logs for a task instance.	`dag_id` (`string`, required), `airflow_run_id` (`string`, required), `task_id` (`string`, required), `try_number` (`integer \| null`, default `null`), `full_content` (`boolean`, default `true`)
`airflow_dag_source`	Fetch DAG source from the API and optionally fall back to local DAG files.	`dag_id` (`string`, required), `use_local_fallback` (`boolean`, default `true`)

Control

Tool	Purpose	Parameters
`airflow_pause_dag`	Pause or unpause a DAG.	`dag_id` (`string`, required), `is_paused` (`boolean \| null`, default `null`)
`airflow_trigger`	Trigger a DAG run and return the run URL when available.	`dag_id` (`string`, required), `conf` (`object \| null`, default `null`), `airflow_run_id` (`string \| null`, default `null`), `logical_date` (`string \| null`, default `null`), `connection_id` (`string \| null`, default `null`)

Usage Examples

# Verify the default saved Airflow connection
airflow_config(action="TEST")

# List active DAGs from the default connection
airflow_list_dags()

# Review recent failed runs across all DAGs on a named connection
airflow_get_all_dag_runs(
    connection_id="production",
    state="failed",
    limit=25,
    order_by="-logical_date",
)

# Trigger a DAG with runtime configuration
airflow_trigger(
    dag_id="daily_orders",
    conf={"requested_by": "genesis", "mode": "backfill"},
)

# Fetch task logs for a specific run
airflow_get_task_logs(
    dag_id="daily_orders",
    airflow_run_id="manual__2026-04-23T12:00:00+00:00",
    task_id="load_orders",
    full_content=False,
)

Notes and Limitations

airflow_config(action="GET") returns the saved deployment URL, username, home, and DAG path for classic Airflow connections, and masks the password in the response. Astronomer connections expose the deployment URL only (no username/password fields). Amazon MWAA connections expose the environment name and region, never the AWS credentials.
Tools that accept connection_id resolve airflow.<name> against Airflow connections, astronomer.<name> against Astronomer connections, and mwaa.<name> against Amazon MWAA connections. A bare name (for example production) resolves against Airflow connections only. When connection_id is omitted, Genesis uses the caller’s single scoped default among the three types. If more than one type has a scoped default, Genesis returns an error until you pass connection_id.
On Amazon MWAA, task logs come from the CloudWatch Logs group airflow-<environment-name>-Task, and airflow_get_failed_tasks is one query returning at most 100 task instances. Results carry no Airflow UI links.
start_date, end_date, and logical_date should be ISO 8601 strings.
airflow_dag_source tries the Airflow API first. When local fallback is enabled, it looks in dags_path or home/dags.
If the API has no source and local fallback is disabled or finds no file, airflow_dag_source returns an unsuccessful result with any available source pointers: filepath, link, and note.
DAG and run inspection tools add dag_url, run_url, or both when Genesis can derive the Airflow UI link from the saved connection and version.
Tools that do not expose connection_id, such as airflow_config and airflow_dag_source, use the caller’s single scoped default among Airflow, Astronomer, and Amazon MWAA connections.

Troubleshooting

No Airflow connection configured

Open Config -> Connections, choose Data Pipelines, and create an Airflow, Astronomer, or Amazon MWAA connection.
Set a user or role default on one saved connection if you want the tool group to work without connection_id.
If an in-chat error says Settings -> Connections, use the same Connections page shown in the Config UI.

More than one Airflow-type scoped default exists

Clear the user or role default on the extra connections, or pass an explicit connection_id such as airflow.production, astronomer.production, or mwaa.production.
Genesis does not pick between two scoped defaults automatically.

Saved connection exists but Airflow tools still fail

For classic Airflow, confirm host, username, password, and version are populated.
For Astronomer, confirm host (deployment URL) and api_token are populated and version matches the deployment (2 or 3).
For Amazon MWAA, an “access denied” error means the IAM identity lacks airflow:InvokeRestApi or airflow:CreateWebLoginToken on the environment’s Airflow role; a task-log error that names logs:GetLogEvents means it cannot read the airflow-<environment-name>-Task log group. “Was not found” means the environment name or region is wrong.
Run Test Connection in the Connections UI, then retry airflow_config(action="TEST").
If your password or API token comes from a secret reference, verify the acting user and agent can resolve that secret.

Explicit connection_id not found

Pass the saved connection name with a type prefix, for example airflow.production, astronomer.production, or mwaa.production.
A bare name such as production resolves only against Airflow connections.
Legacy shared Airflow entries are ignored by the current Airflow tools.

DAG source fallback does not find files

Set DAGs Path on the saved Airflow connection when Genesis can access the DAG files directly.
If you leave DAGs Path blank, set Airflow Home so Genesis can try home/dags.
If Genesis cannot reach the Airflow host filesystem, use the API-backed source path instead of local fallback.

Task logs are incomplete or unavailable

Confirm the task has already started or completed in Airflow.
Retry with full_content=True when you need the full log body.
If the task retried, pass the matching try_number to inspect the same attempt shown in the Airflow UI.

Getting Started

Getting to Know Genesis

Creating A Mission

User Guide

Blueprint

Genesis Data Agents

Genesis Prover

Genesis Apps

Genesis REST API

Genesis Agent Security Controls

Genesis Data Agent's Toolkit

Setup

Slack and Teams

Data Connections

Deployment Options

Overview

Tool Name

Purpose

Connection Requirements

Classic Airflow (username/password)

Astronomer Cloud (API token)

Amazon MWAA (IAM)

Function Reference

Connection and Discovery

Run and Failure Inspection

Task Inspection

Control

Usage Examples

Notes and Limitations

Troubleshooting

​Overview

​Tool Name

​Purpose

​Connection Requirements

​Classic Airflow (username/password)

​Astronomer Cloud (API token)

​Amazon MWAA (IAM)

​Function Reference

​Connection and Discovery

​Run and Failure Inspection

​Task Inspection

​Control

​Usage Examples

​Notes and Limitations

​Troubleshooting

​Related Pages

Overview

Tool Name

Purpose

Connection Requirements

Classic Airflow (username/password)

Astronomer Cloud (API token)

Amazon MWAA (IAM)

Function Reference

Connection and Discovery

Run and Failure Inspection

Task Inspection

Control

Usage Examples

Notes and Limitations

Troubleshooting

Related Pages