airflow-list-runs

Instructions

Implementation Reference

• src/tools/dags.ts:53-82 (handler)

  The main handler function for the 'airflow-list-runs' tool. Calls Airflow REST API v2 endpoint /dags/{dagId}/dagRuns with optional state filter and orders by -logical_date.

  export async function airflowListRuns(args: z.infer<typeof airflowListRunsSchema>): Promise<unknown> {
    const qs = new URLSearchParams();
    qs.set("limit", String(args.limit));
    // Airflow 3.x v2 uses logical_date (formerly execution_date in v1)
    qs.set("order_by", "-logical_date");
    if (args.state) qs.append("state", args.state);
    const data = await airflowFetch<{ dag_runs: AirflowDagRun[]; total_entries: number }>(
      `/dags/${encodeURIComponent(args.dagId)}/dagRuns?${qs.toString()}`,
    );
    return {
      dagId: args.dagId,
      totalEntries: data.total_entries,
      count: data.dag_runs.length,
      runs: data.dag_runs.map((r) => ({
        dagRunId: r.dag_run_id,
        state: r.state,
        logicalDate: r.logical_date ?? r.execution_date,
        runType: r.run_type,
        startDate: r.start_date,
        endDate: r.end_date,
        duration: r.duration,
        queuedAt: r.queued_at,
        runAfter: r.run_after,
        triggeredBy: r.triggered_by,
        triggeringUser: r.triggering_user_name,
        conf: r.conf,
        note: r.note,
      })),
    };
  }

• src/tools/dags.ts:47-51 (schema)

  The Zod input schema for the 'airflow-list-runs' tool. Defines parameters: dagId (required string), state (optional filter string), limit (number 1-200, default 20).

  export const airflowListRunsSchema = z.object({
    dagId: z.string().describe("Airflow DAG id"),
    state: z.string().optional().describe("Filter by state (success | running | failed | queued | up_for_retry | ...)"),
    limit: z.coerce.number().int().min(1).max(200).default(20),
  });

• src/index.ts:47-47 (registration)

  Registration of the 'airflow-list-runs' tool on the MCP server via the helper function `tool()`. Links the schema (airflowListRunsSchema.shape) and handler (wrapToolHandler(airflowListRuns)).

  tool("airflow-list-runs", "List recent runs of one Airflow DAG, optionally filtered by state, ordered newest first", airflowListRunsSchema.shape, wrapToolHandler(airflowListRuns));

• src/index.ts:36-41 (helper)

  The `tool()` helper function that wraps registration: it calls registry.register() and conditionally registers with the MCP server based on category enablement.

  function tool(name: string, description: string, schema: any, handler: any): void {
    registry.register(name, description, currentCategory);
    if (registry.isEnabled(currentCategory)) {
      server.tool(name, description, schema, handler);
    }
  }

• src/tools/utils.ts:28-52 (helper)

  The wrapToolHandler helper that wraps the handler with error handling (redaction patterns and error extractors for WriteBlockedError and AirflowApiError).

  export const wrapToolHandler = createWrapToolHandler({
    redactionPatterns: [
      /AIRFLOW_PASSWORD/i,
      /Authorization:\s*Basic\s+[A-Za-z0-9+/=]+/i,
    ],
    errorExtractors: [
      {
        match: (error) => error instanceof WriteBlockedError,
        extract: (error) => ({
          kind: "passthrough",
          text: (error as WriteBlockedError).message,
        }),
      },
      {
        match: (error) => error instanceof AirflowApiError,
        extract: (error) => {
          const err = error as AirflowApiError;
          return {
            kind: "structured",
            data: { message: err.message, status: err.status, details: err.body },
          };
        },
      },
    ],
  });