Approve an agent’s tool calls¶

What each decision does¶

Approve runs the call and the agent carries on from where it stopped.

Reject does not fail the task. The agent is told the call was denied – with the reviewer’s reason when one is given, such as “Refunds over $40 need a support ticket number.” – and carries on without it, so it can explain the refusal, ask for what is missing, or take another route.

While it waits, the task is in the awaiting_input state and holds no worker slot. When the model calls several tools in one step, the ones that need approval are decided together and the others run straight away, once.

usage_limits covers both sides of the pause, so a cost_limit is not reset by it.

An agent that needs approval more than once¶

An agent usually decides its next call after it sees the result of the last one, so a refund agent may ask to refund one order, read the result, and then ask to refund another. On Airflow 3.4 and later, each step that calls a marked tool pauses the task again, and the reviewer sees only that step’s calls. usage_limits counts across every pause, so a run that pauses many times can reach request_limit (50 by default) partway through, after earlier approved calls have run.

Each request is numbered in its subject, and the review form sends the number back with the answer in a hidden field. An answer from a form left open on an earlier request, such as a second reviewer who never reloaded the page, is not applied to the calls waiting now: the task asks about them again under a new number. An answer sent through the REST API without that field is applied to the waiting request.

A retry or a clear asks again for the calls the new attempt makes, including ones a reviewer approved in an earlier attempt that then failed. The approved call already ran once, so keep gated tools idempotent.

On Airflow 3.3 a task instance can ask only once per Dag run, retries and clears included. Airflow 3.3 keeps the first request’s subject and body when the same task instance asks again, so a second request would show the reviewer the earlier call while asking about the new one. The task fails with ToolApprovalAlreadyRequestedError instead, and does not retry. On 3.3, have the agent request the gated calls in one step, or give each irreversible action its own task.

Timeouts and who decides¶

tool_approval_timeout bounds the pause; None (the default) waits for as long as it takes. What a timeout does is set by on_tool_approval_timeout:

"fail" (default) fails the task.
"deny" rejects the pending calls, and the agent carries on without them. The agent is told nobody answered in time, not that a person refused. "deny" needs a tool_approval_timeout.

There is no approve-on-timeout: a pause that approves itself when nobody answers guards nothing.

tool_approval_assigned_users limits who may decide, with the same user list require_approval on the LLM operators takes (see Approval gates for LLM operators).

The arguments shown to the reviewer pass through Airflow’s secrets masker first, so a value under a key such as api_key or password, or a value already registered as a secret, is shown masked.

Let a decision model approve the routine calls¶

Most refunds are routine, and a person who approves every one of them stops reading them. A decision model can answer “may this call run without a person?” for each one, and pydantic-ai’s HandleDeferredToolCalls capability is the place to ask: its handler sees every call waiting for approval, with its validated arguments, before the task pauses. It approves the calls the model is sure about, denies the ones the model is sure should not run, and leaves the rest. The task pauses on whatever it leaves, exactly as above.

A limit you can write down, such as “refunds over $50”, needs no model: the .approval_required() function above can check the arguments itself. The model earns its call on the question a rule cannot answer, whether the call fits the request: the right order, an amount the customer’s problem supports, an action somebody asked for.

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

ORDERS = {7: 42.00}


def lookup_order(order_id: int) -> str:
    """Look up an order."""
    return f"order {order_id}: ${ORDERS[order_id]:.2f}, carrier marked it lost"


def refund_order(order_id: int, amount: float, reason: str) -> str:
    """Refund an order. Irreversible: the money leaves the account."""
    return f"refunded ${amount:.2f} on order {order_id}"


def within_order_total(ctx: RunContext, order_id: int, amount: float, reason: str) -> None:
    """A rule, not a judgment: a refund it refuses never reaches the model or a person."""
    if order_id not in ORDERS:
        raise ModelRetry(f"There is no order {order_id}.")
    if amount > ORDERS[order_id]:
        raise ModelRetry(f"Order {order_id} cost ${ORDERS[order_id]:.2f}; refund at most that.")


shop = FunctionToolset(
    tools=[
        lookup_order,
        Tool(refund_order, requires_approval=True, args_validator=within_order_total),
    ]
)
TOOL_DESCRIPTIONS = {tool.__name__: tool.__doc__ for tool in (lookup_order, refund_order)}


class Verdict(UseEnumMemberDocstrings, str, Enum):
    """Whether a proposed tool call may run without a person checking it first."""

    allow = "allow"
    """The call does what the request asks, on the right record, at a size the request supports."""

    review = "review"
    """The call may be right, but a person should check it: an unusual amount, an irreversible step the request does not clearly ask for, or context that is missing."""

    deny = "deny"
    """The call does not match the request: the wrong record, an amount far beyond what was asked, or an action nobody asked for."""


# Below this, the answer is not acted on and the call goes to a person.
MIN_CONFIDENCE = 0.7


def customer_request(ctx: RunContext) -> str:
    """The request the agent is working on. A run resumed after a pause has no prompt, only history."""
    if isinstance(ctx.prompt, str):
        return ctx.prompt
    prompts = [
        part.content
        for message in ctx.messages
        for part in message.parts
        if isinstance(part, UserPromptPart) and isinstance(part.content, str)
    ]
    return prompts[-1] if prompts else ""


def refused_earlier(ctx: RunContext, tool_name: str) -> bool:
    """Whether a person or the model refused a call to this tool earlier in the run."""
    return any(
        isinstance(part, ToolReturnPart) and part.tool_name == tool_name and part.outcome == "denied"
        for message in ctx.messages
        for part in message.parts
    )


async def judge_tool_calls(ctx: RunContext, requests: DeferredToolRequests) -> DeferredToolResults:
    """Approve or deny the calls the decision model is sure about; leave the rest for a person."""
    judge = PydanticAIHook(llm_conn_id="decision_default").create_agent(
        output_type=Verdict,
        instructions=(
            "A support agent proposed the tool call below while handling the customer request. "
            "Decide whether it may run without a person checking it first."
        ),
    )
    results = DeferredToolResults()
    for call in requests.approvals:
        if refused_earlier(ctx, call.tool_name):
            # The model does not see why the earlier call was refused, so it must not overrule that.
            log.info(
                "%s (%s) was refused earlier in this run; asking a person", call.tool_name, call.tool_call_id
            )
            continue
        question = (
            f"Customer request:\n{customer_request(ctx)}\n\n"
            f"Proposed tool call: {call.tool_name}\n"
            f"What the tool does: {TOOL_DESCRIPTIONS.get(call.tool_name, 'unknown')}\n"
            f"Arguments: {json.dumps(call.args_as_dict())}"
        )
        try:
            answer = await judge.run(question)
        except Exception:
            # An unreachable judge is not a verdict: leave the call for a person.
            log.warning("Judge failed on %s; asking a person", call.tool_name, exc_info=True)
            continue
        confidence = ModelConfidence.from_result(answer).confidence.get("response")
        acted_on = confidence is not None and confidence >= MIN_CONFIDENCE
        # No arguments here: they can carry secrets, which the review page masks and a log line would not.
        log.info(
            "Judge on %s (%s): %s at confidence %s, bar %s; %s",
            call.tool_name,
            call.tool_call_id,
            answer.output.value,
            confidence,
            MIN_CONFIDENCE,
            "acting on it" if acted_on and answer.output is not Verdict.review else "asking a person",
        )
        if not acted_on:
            continue
        if answer.output is Verdict.allow:
            results.approvals[call.tool_call_id] = True
        elif answer.output is Verdict.deny:
            results.approvals[call.tool_call_id] = ToolDenied("An automated check refused this call.")
    return results


@dag(tags=["example"])
def example_agent_tool_judge():
    AgentOperator(
        task_id="handle_ticket",
        llm_conn_id="pydanticai_default",
        prompt="The customer says order 7 never arrived. Resolve it.",
        toolsets=[shop],
        capabilities=[HandleDeferredToolCalls(handler=judge_tool_calls)],
        tool_approval_timeout=timedelta(hours=4),
    )

What happens to each call:

The model answers	Confidence at or above `MIN_CONFIDENCE`	Below it, or none reported
`allow`	The call runs.	A person decides.
`deny`	The call does not run; the agent is told, as after a Reject.	A person decides.
`review`	A person decides.	A person decides.
The model call raises	A person decides.	A person decides.

Once a person approves, the call runs without the model being asked again. Once a call to a tool has been refused, by a person or by the model, later calls to that tool in the same run go to a person: the model does not see why the earlier call was refused, so it does not get to overrule that.

Keep hard limits in the tool. A decision model reads a tool argument as data, not as an attempt to manipulate it, so an argument written to argue for its own approval can move the answer (see Decision models). within_order_total is the tool’s args_validator: it runs in the worker after the arguments validate and before the call is marked for approval, so a refund above the order total goes back to the agent without the model or a person being asked about it.

Set the bar from your own calls. Run the handler over calls your agent actually makes, and raise MIN_CONFIDENCE until the calls it lets through are ones you would have approved. A model that hesitates sends more calls to a person, which costs time; one that is trusted too far costs whatever the tool does.

The Required Actions page shows the call, not the model’s answer; the handler logs the answer and its confidence to the task log, where the reviewer can look. A call the handler leaves for a person follows every rule on this page, including how often one task instance can ask in a Dag run. Where tool approval is not available, such a call fails the task. An exception the handler does not catch, such as a missing decision_default connection, fails the task as well.

Requirements and limits¶

Airflow 3.3 or later. On older versions a tool marked for approval fails the task, as it did before.
Not together with durable=True, enable_hitl_review=True, a CodeMode capability, or a SandboxToolset that provisions its own sandbox. Each assumes the run finishes in one go; that sandbox, for one, is destroyed when the run pauses. With any of them, a marked tool fails the task, except under code mode, where a marked tool called from run_code does not run and the model may carry on without it (see Code mode). A SandboxToolset attached to a sandbox another task owns keeps its files through the pause, so it is allowed; the wait spends that sandbox’s lifetime (Share a sandbox between tasks).
Tools that hand work to an external system (pydantic-ai’s CallDeferred) are not supported; the task fails without retrying.
The conversation so far, including tool results, is kept in the task’s task state store while the task waits. It is deleted when the resumed run succeeds or fails, and replaced when it pauses again; a later try deletes one left behind by a try that ended while waiting. With a tool_approval_timeout, it also expires a day after the timeout.
If a templated connection id (see Templated connection IDs) renders differently when the task resumes, the task fails rather than run an approved call against a connection the reviewer did not see. The check compares toolset ids, which name the connection for SQLToolset, MCPToolset and HookToolset; it does not notice an edit to the connection itself.
Each resumed run gets its own pydantic-ai run_id, <task-instance id>-resumed-<n> when it resumes from pause n. The run_id XCom holds the last one, and the usage XCom covers every part of the run.