Explain a revenue anomaly¶

Daily revenue moved more than ten percent against its trailing average and someone has to find out which region and channel explain it before the morning stand-up. This Dag gives an agent two tools: read-only queries against two warehouse tables, capped at 500 rows, and a pandas sandbox for the pivots. It returns a typed finding with the suspected cause and a confidence. Airflow injects the run date, keeps the warehouse credential where the model never sees it, and destroys the sandbox when the run ends.

What this demonstrates¶

Agents with tools: AgentOperator and @task.agent – AgentOperator runs a multi-turn agent as one task, with output_type for a typed result.
SQL databases: SQLToolset – SQLToolset exposes named query operations; the model sees rows, never the connection.
Sandboxed execution for agents – SandboxToolset runs everything the model writes off the worker.
Templating – {{ ds }} in the prompt ties the question to the Dag run’s date.

Run it¶

Install the provider with the SQL and Modal extras and authenticate with Modal:

pip install "apache-airflow-providers-common-ai[openai,sql,modal]"
modal setup

Create a database connection named warehouse that has analytics.daily_revenue and analytics.orders tables.

Trigger the Dag for a date:

airflow dags test example_sandbox_agent_investigation 2026-03-01

The investigate task log shows every SQL call and every script the agent ran in the sandbox, and its XCom holds the Findings record.

The Dag¶

airflow/providers/common/ai/example_dags/example_sandbox_toolset.py[source]

    # Packages come from the image, so the sandbox needs no network at all. The
    # image is built by Modal the first time it is used and cached after that.
    ANALYSIS_IMAGE = modal.Image.from_registry("python:3.12-slim").pip_install("pandas")

    @dag(
        schedule=None,
        start_date=datetime(2024, 1, 1, tzinfo=timezone.utc),
        catchup=False,
        tags=["example", "sandbox"],
    )
    def example_sandbox_agent_investigation():
        """Work out why daily revenue moved, with the warehouse and the workspace kept apart."""
        AgentOperator(
            task_id="investigate",
            prompt=(
                "Revenue for {{ ds }} moved more than 10% against the trailing seven-day "
                "average. Find out which region and channel explain the move, and how sure you are."
            ),
            system_prompt=(
                "You are a revenue analyst. Use the SQL tools to pull daily aggregates by region "
                "and channel for the two weeks up to and including the date in question. Write "
                "the rows into the sandbox as CSV and use pandas there for pivots and "
                "comparisons rather than doing arithmetic in your head. The sandbox has no "
                "network access and pandas is already installed; do not try to install anything. "
                "If a script fails, read the traceback, fix it and run it again."
            ),
            llm_conn_id="pydanticai_default",
            output_type=Findings,
            toolsets=[
                # The warehouse credential is held by the hook, in the task process. The
                # model calls named operations and sees rows; it never sees the password
                # and cannot pass it into the sandbox except by typing rows it was shown.
                SQLToolset(
                    db_conn_id="warehouse",
                    allowed_tables=["analytics.daily_revenue", "analytics.orders"],
                    max_rows=500,
                ),
                # Everything the model writes and runs happens here: off the worker, no
                # egress, nothing of Airflow's inside, destroyed when the run ends.
                SandboxToolset(
                    ModalSandboxBackend(image=ANALYSIS_IMAGE, cpu=1),
                    # The default spec already denies all egress and injects no
                    # environment. Stated here so the intent is visible in the Dag.
                    spec=SandboxSpec(block_network=True),
                ),
            ],
        )

Adapting it¶

Trigger it from a data-quality check that fires when the move exceeds your threshold, so the agent runs only on days that need explaining.
Swap ModalSandboxBackend for SbxSandboxBackend to run locally; the same file has that variant (see Sandbox backends).
Add an Approval gates for LLM operators step before the finding reaches the finance channel.
Widen allowed_tables only as far as the question needs. The limit is what makes the agent safe to run unattended.