Airflow AI Provider 1.0: Agents, Sandboxes, and Human Approval in Production DAGs

Posted on Oct 21 • Originally published at mech.app

Apache Airflow just shipped AI Provider 1.0, bringing agent orchestration primitives to a workflow engine already running in thousands of production data pipelines. This is not a new agent framework. It is a set of DAG tasks that let you run LLM agents inside existing Airflow workflows, with sandboxing, human approval gates, and retry logic that understands when a model fails versus when a tool call fails.

The provider installs on Airflow 2.11 and later, so teams still on Airflow 2 can start using agents without upgrading the entire platform. The core primitives are agent tasks, sandboxed code execution, and human-in-the-loop approval gates that integrate with Airflow's XCom system for passing context between tasks.

The Common AI provider ( apache-airflow-providers-common-ai ) adds three main capabilities to Airflow DAGs:

An agent task is a model working in a loop. It calls tools (functions the DAG author allows), reads results, and picks the next step until it has an answer. The provider wraps this loop in Airflow's TaskFlow API, so it behaves like any other DAG task with retries, logging, and XCom data passing.

The provider supports OpenAI, Anthropic, Google, and Bedrock models through a unified connection type called pydanticai. You configure the model and API key in an Airflow connection, and the agent task reads it at runtime.

Airflow DAGs are directed acyclic graphs where each node is a task. The AI Provider adds agent tasks that fit into this topology like any other task. The difference is what happens inside the task.

A traditional Airflow task runs a function once. An agent task runs a loop:

The agent task returns its final answer through XCom, so downstream tasks can use it. If the agent fails (model timeout, invalid tool call, sandbox error), Airflow's retry logic kicks in. The provider lets you configure whether retries happen automatically or only after a model decides the failure is transient.

When an agent writes code, the provider runs it in a sandbox instead of the Airflow worker process. This prevents agent-generated code from:

The sandbox is a separate process with limited filesystem access and no access to Airflow's internal state. The agent task passes code to the sandbox, waits for the result, and returns it to the model. If the sandbox crashes or times out, the agent task fails and Airflow retries it.

The sandbox does not use containers or VMs. It is a Python subprocess with restricted imports and filesystem access. This keeps overhead low but means you need to trust the sandbox implementation to enforce boundaries. For higher isolation, you would wrap the agent task in a KubernetesPodOperator or similar.

On Airflow 3.3 and later, agent tasks can pause and wait for human approval before executing a tool call. This is useful when the agent wants to post to Slack, send an email, or run a write query.

The approval gate works through Airflow's existing approval mechanism. When the agent calls a tool marked as requiring approval, the task pauses and creates an approval request. A person reviews the request in the Airflow UI, approves or rejects it, and the agent task resumes.

If the person rejects the tool call, the agent task fails and Airflow retries it (or not, depending on retry configuration). If approved, the tool executes and the result flows back to the model.

This integrates with Airflow's XCom system. The approval request is an XCom value that downstream tasks can read. The approval decision is another XCom value that the agent task reads to continue or fail.

Airflow retries failed tasks automatically, but agent tasks need smarter retry logic. A model timeout is transient. An invalid SQL query is not.

Automatic retries treat every failure as transient. If the model times out, the tool call fails, or the sandbox crashes, Airflow retries the entire agent task. This works for flaky APIs or rate limits but wastes tokens on persistent errors.

Model-decided retries let the model decide whether to retry. When a tool call fails, the error message flows back to the model. The model can:

This mode requires the model to understand error messages and make retry decisions. It saves tokens on persistent errors but adds latency because the model has to process the error and decide.

You configure retry mode per agent task. The default is automatic retries with a max retry count.

The provider exposes metrics for each agent task:

These metrics flow into Airflow's existing observability stack. You can query them through Airflow's REST API, export them to Prometheus, or view them in the task logs.

The provider also logs every tool call with arguments and results. This is useful for debugging why an agent made a particular decision or why a tool call failed. The logs include:

For token usage tracking, the provider reads token counts from the model API response. If the model does not expose token counts, the provider estimates them using a tokenizer.

Here is an agent task that queries a warehouse to explain a revenue drop:

from airflow.decorators import dag, task from datetime import datetime from airflow.providers.common.ai.decorators import agent_task @dag ( start_date = datetime ( 2026, 10, 1 ), schedule = None ) def revenue_analysis (): @agent_task ( conn_id = " pydanticai_default ", tools = [ " sql_query " ], sandbox = True, require_approval = False ) def explain_revenue_drop ( context ): """ Query the warehouse to explain why revenue dropped 15% last week. You have read-only access to the sales and customer tables. """ return { " prompt ": " Revenue dropped 15% last week.

Query the warehouse to find the cause. ", " max_iterations ": 10 } @task def post_to_slack ( explanation: str ): # Post the explanation to Slack pass result = explain_revenue_drop () post_to_slack ( result ) revenue_analysis () Enter fullscreen mode Exit fullscreen mode The agent_task decorator wraps the function in an agent loop. The tools parameter lists available tools (in this case, a SQL query tool).

The sandbox=True parameter runs agent-generated code in a sandbox. The require_approval=False parameter allows the agent to execute tool calls without human approval.

The agent task returns a dictionary with the prompt and max iterations. The provider runs the agent loop, calls tools as needed, and returns the final answer through XCom. The downstream post_to_slack task receives the explanation and posts it.

Airflow adds scheduling overhead. If your agent workflow is latency-sensitive (sub-second responses), Airflow is not the right tool. If your agent workflow runs on a schedule or as part of a larger data pipeline, Airflow makes sense.

The sandbox isolation is weaker than containers or VMs. If you need strong isolation (untrusted code, compliance requirements), wrap the agent task in a KubernetesPodOperator or run it in a separate cluster.

Human approval gates are useful for high-risk actions (posting to Slack, running write queries) but add latency. If your agent workflow is fully automated, disable approval gates.

Model timeouts: The provider sets a default timeout of 60 seconds per model call. If the model does not respond, the task fails and Airflow retries it. You can increase the timeout in the agent task configuration.

Invalid tool calls: If the model calls a tool with invalid arguments, the tool returns an error. In automatic retry mode, Airflow retries the entire task. In model-decided retry mode, the error flows back to the model, which can retry with different arguments.

Sandbox crashes: If the sandbox process crashes, the agent task fails. Airflow retries it. The sandbox does not persist state between retries, so the agent starts from scratch.

Approval timeouts: If a human does not approve a tool call within the timeout (default 1 hour), the agent task fails. You can configure the timeout per task.

Token limit exhaustion: If the agent loop uses too many tokens, the model API returns an error. The provider does not track cumulative token usage across retries, so you need to monitor this separately.

The provider installs on Airflow 2.11, but some features require Airflow 3.3:

On Airflow 2.11, change imports from airflow.decorators to airflow.operators.python. The agent_task decorator is the same, but the import path is different.

The provider is useful for teams that already run Airflow and want to add agent tasks without deploying a separate agent framework. It is not a replacement for standalone agent frameworks like LangGraph or CrewAI, but it fills a gap for teams that want to run agents inside existing workflow orchestration.