af tasks: add a list command for task instances (cross-DAG / wildcard)

[RobGoretsky]

Intro

Thanks again for astro-airflow-mcp - it's become the backbone of the on-call Airflow Skill we're building, and the af CLI abstraction has let us move fast. Raising one more gap we ran into, and happy to send a PR if the shape sounds right to you.

What we're trying to do

Our Skill does a lot of fleet-wide triage where we don't know the DAG or run ID up front, we're asking a question across everything. A few of our routine queries:

• "Show me every task instance that failed in the last hour, across all DAGs" (war-room triage)
• "What's running right now, fleet-wide, and how long has it been going?"
• "For this one DAG, list the task instances across its recent runs" (catch-up recovery monitoring)

All of these map cleanly to the list endpoint, which accepts ~ for both dag_id and dag_run_id:

GET /api/v2/dags/~/dagRuns/~/taskInstances?state=failed&end_date_gte=2026-06-10T15:00:00Z

The limitation

There's no af tasks subcommand that reaches this list endpoint. The existing commands cover other needs:

• af tasks list <dag_id> lists task definitions (dags/{dag_id}/tasks), not runtime instances
• af tasks instance <dag_id> <dag_run_id> <task_id> is a single instance and needs all three IDs

There is an adapter method that hits the list endpoint, get_task_instances (adapters/airflow_v3.py:363), but it takes dag_id/dag_run_id as required positionals, exposes no filters, and isn't wired to any CLI command (it's only called internally by diagnose and trigger-wait):

def get_task_instances(
    self, dag_id: str, dag_run_id: str, limit: int = 100, offset: int = 0
) -> dict[str, Any]:
    ...
    return self._call(
        f"dags/{dag_id}/dagRuns/{dag_run_id}/taskInstances",
        params={"limit": limit, "offset": offset},
    )

So today we drop to the escape hatch for every one of these:

af api "dags/~/dagRuns/~/taskInstances" -F state=failed -F limit=100

It works, but it skips the typed CLI surface that makes the rest of our Skill nice to build on.

Proposed fix

The natural home looks like a new list_task_instances adapter method that mirrors list_dag_runs (adapters/airflow_v3.py:135), rather than overloading the existing get_task_instances, which keeps its current "required IDs" contract for its internal callers. list_dag_runs already establishes the pattern of defaulting the path segment to ~ and forwarding arbitrary filters via **kwargs, so we'd do the same here for our proposed new list_task_instances:

def list_task_instances(
    self,
    dag_id: str | None = None,
    dag_run_id: str | None = None,
    limit: int = 100,
    offset: int = 0,
    **kwargs: Any,
) -> dict[str, Any]:
    """List task instances. Use '~' or None for dag_id/dag_run_id to span all."""
    dag_id_param = dag_id if dag_id else "~"
    run_param = dag_run_id if dag_run_id else "~"
    return self._call(
        f"dags/{dag_id_param}/dagRuns/{run_param}/taskInstances",
        params={"limit": limit, "offset": offset},
        **kwargs,
    )

Then a CLI command modeled on list_dag_runs in cli/runs.py:

@app.command("instances")
def list_task_instances(
    dag_id: Annotated[str | None, typer.Option("--dag-id", "-d", help="Filter by DAG ID (default: all DAGs)")] = None,
    dag_run_id: Annotated[str | None, typer.Option("--run-id", "-r", help="Filter by DAG run ID (default: all runs)")] = None,
    task_id: Annotated[str | None, typer.Option("--task-id", "-t", help="Filter by task ID")] = None,
    state: Annotated[str | None, typer.Option("--state", "-s", help="Filter by state: running, success, failed, ...")] = None,
    start_date_gte: Annotated[str | None, typer.Option("--start-date-gte", help="Instances with start_date >= value")] = None,
    start_date_lte: Annotated[str | None, typer.Option("--start-date-lte", help="Instances with start_date <= value")] = None,
    end_date_gte: Annotated[str | None, typer.Option("--end-date-gte", help="Instances with end_date >= value")] = None,
    end_date_lte: Annotated[str | None, typer.Option("--end-date-lte", help="Instances with end_date <= value")] = None,
    duration_gte: Annotated[float | None, typer.Option("--duration-gte", help="Instances with duration >= seconds")] = None,
    duration_lte: Annotated[float | None, typer.Option("--duration-lte", help="Instances with duration <= seconds")] = None,
    try_number: Annotated[int | None, typer.Option("--try", help="Filter by try number")] = None,
    pool: Annotated[str | None, typer.Option("--pool", help="Filter by pool")] = None,
    operator: Annotated[str | None, typer.Option("--operator", help="Filter by operator")] = None,
    order_by: Annotated[str | None, typer.Option("--order-by", help="Sort field; '-' prefix for descending")] = None,
    limit: Annotated[int, typer.Option("--limit", "-l")] = 100,
    offset: Annotated[int, typer.Option("--offset", "-o")] = 0,
) -> None:
    """List task instances across runs/DAGs (dags/~/dagRuns/~/taskInstances)."""
    try:
        kwargs: dict[str, Any] = {}
        for key, val in (
            ("task_id", task_id), ("state", state),
            ("start_date_gte", start_date_gte), ("start_date_lte", start_date_lte),
            ("end_date_gte", end_date_gte), ("end_date_lte", end_date_lte),
            ("duration_gte", duration_gte), ("duration_lte", duration_lte),
            ("try_number", try_number), ("pool", pool), ("operator", operator),
            ("order_by", order_by),
        ):
            if val is not None:
                kwargs[key] = val

        adapter = get_adapter()
        data = adapter.list_task_instances(
            dag_id=dag_id, dag_run_id=dag_run_id, limit=limit, offset=offset, **kwargs
        )
        if "task_instances" in data:
            output_json(wrap_list_response(data["task_instances"], "task_instances", data))
        else:
            output_json({"message": "No task instances found", "response": data})
    except Exception as e:
        output_error(str(e))

We picked this set by looking at the query params the dags/{dag_id}/dagRuns/{dag_run_id}/taskInstances list endpoint actually exposes in the Airflow 3 spec, and pulling the ones that map to common triage questions (recent failures, slow tasks, retries). The endpoint has more if you want them (logical_date_*, run_after_*, updated_at_* ranges; queue, executor, map_index, version_number; task_group_id; task_display_name_pattern), we just left those out to keep the surface focused.

One thing worth flagging: several of these (state, try_number, pool, operator, order_by, ...) are declared as array params on this endpoint. We've scoped them as single-value options above for simplicity, but exposing the array form depends on the repeated-param handling in #223. So #223 and this are complementary, that fix is what would let --try 2 --try 3 work here.

A note on naming

Naming is the only real puzzle here, so to save you the archaeology: the CLI command name looks decoupled from the adapter method name (list_asset_events is registered as af assets events, not af assets list), so we'd suggest treating them as two separate choices:

• Adapter method: list_task_instances, parallel to list_dag_runs and list_asset_events. No collision there.
• CLI command: assets is the precedent that already hit our exact situation. af assets list was taken by the primary resource, so the secondary list-able sub-resource became af assets events (a distinct command under the same group, not a new group). af tasks list is likewise already task definitions, so af tasks instances follows that pattern for the task-instance sub-resource.

The one wrinkle: af tasks instances (list) would sit one keystroke from the existing af tasks instance (get one). The assets group avoided that by using a different word (events) rather than a singular/plural pair, but there's no obvious alternate word for task instances, so the plural seems like the least-surprising option. Flagging in case you'd rather disambiguate.

A couple of other things we'd defer to you:

1. How far to go on the param set, the focused list above vs. exposing the full endpoint surface.
2. Whether you'd want the Airflow 2 adapter updated in the same PR for parity.

Glad to open a PR along these lines if you're up for it. Thanks again.

Environment

• af CLI: 0.8.2
• Airflow target: 3.x (/api/v2)