Add human approval for agent tool calls

[kaxil]

An agent that can look things up freely should not refund an order, send an email or write to a production table without a person seeing the call first. Mark those tools with pydantic-ai's own approval API, and AgentOperator / @task.agent pause the task in front of the call and ask on the Required Actions page, with the tool name and its arguments:

shop = FunctionToolset(tools=[lookup_order, refund_order]).approval_required(
    lambda ctx, tool_def, args: tool_def.name == "refund_order"
)

AgentOperator(
    task_id="handle_ticket",
    llm_conn_id="pydanticai_default",
    prompt="The customer says order 7 never arrived. Resolve it.",
    toolsets=[shop],
    tool_approval_timeout=timedelta(hours=4),
)

Approve runs the call and the agent carries on. Reject does not fail the task: the agent gets the reviewer's reason as the tool result and carries on without the call. While it waits, the task is in awaiting_input and holds no worker slot. Needs Airflow 3.3+.

Design rationale

Why not require_approval or enable_hitl_review? Both review an output: require_approval gates an LLMOperator result after the run, and enable_hitl_review reviews an agent's final answer while polling in the worker. Neither can stop a side effect before it happens, which is what a tool call needs.

Marking is pydantic-ai's; Airflow supplies the backend. toolset.approval_required() and Tool(..., requires_approval=True) already exist and make pydantic-ai end the run with DeferredToolRequests. The operator adds DeferredToolRequests to the output type only while building the agent. pydantic-ai strips it from the schema the model sees, and self.output_type (part of the serialized Dag) is unchanged. The pause is the same TaskAwaitingInput path LLMApprovalMixin uses on 3.3+, and the resume continues the run from its transcript with DeferredToolResults.

The transcript goes to the task state store, not the continuation kwargs. It holds tool results such as query rows, which do not belong on the task instance row. The continuation carries the transcript's hash, the pending call ids, the usage so far, and the rendered toolset ids. The resume fails if the transcript changed, or if a templated connection id now renders to a different connection than the reviewer saw.

One approval per task instance per Dag run. Core keeps a single approval request per task instance, across retries and clears, and a repeat request keeps the first one's subject and body (execution_api/routes/hitl.py). A second request would show the reviewer the earlier call while Approve applied to the new one. So the operator records a marker in the task state store, and a second request fails with a non-retryable ToolApprovalAlreadyRequestedError. Allowing several approvals needs core to overwrite the request content on a repeat upsert; that is a separate change.

No approve-on-timeout. on_tool_approval_timeout is "fail" (default) or "deny", and "deny" tells the agent that nobody answered, not that a person refused. usage_limits covers both sides of the pause, so a cost_limit is not reset by it.

Screenshots

On a real scheduler and API server with a test model: two runs waiting on the same tool, answered through the UI.

Approve with the reason left empty: the refund runs and the agent finishes. The resumed run's run_id carries the -resumed suffix.

Reject with a reason: the agent receives it as the tool result, and the task still succeeds.

Gotchas

• Not available together with durable=True, enable_hitl_review=True, code_mode=True or a SandboxToolset, each of which assumes the run finishes in one go. There, a marked tool fails the task as it does today, and adding DeferredToolRequests to output_type by hand raises UnsupportedToolDeferralError.
• Tools that defer to external execution (pydantic-ai's CallDeferred) are not supported and fail without retrying.
• A task whose approved call ran and that then failed cannot ask again in the same Dag run; the retry fails at the gate. The docs say to keep gated tools idempotent and trigger a new run.
• pydantic-ai refuses a run_id already present in the message history, so the resumed run uses <task-instance id>-resumed. The run_id XCom holds that id; usage covers both sides of the pause.
• LoggingToolset now logs a call waiting for approval at INFO rather than as an ERROR "Tool ... failed".
• LLMOperator's assigned-users validation moved into a helper shared with the new tool_approval_assigned_users; its behaviour and messages are unchanged.

---

• Read the Pull Request Guidelines for more information. Note: commit author/co-author name and email in commits become permanently public when merged.
• For fundamental code changes, an Airflow Improvement Proposal (AIP) is needed.
• When adding dependency, check compliance with the ASF 3rd Party License Policy.
• For significant user-facing changes create newsfragment: {pr_number}.significant.rst, in airflow-core/newsfragments. You can add this file in a follow-up commit after the PR is created so you know the PR number.