Agent with human approval

Gate destructive agent tool calls behind human review. The agent pauses before running a flagged tool, an operator approves or rejects via CLI or HTTP, and the workflow resumes exactly where it left off.

Some tool calls shouldn’t run unattended. An agent that can send email, apply a database migration, or deploy to production needs a human checkpoint before any of those run. Flux’s approval primitive — requires_approval on @task.with_options(...) — turns any task into a gate; the agent harness picks up the gate when the task is used as a tool.

This page shows how to wire that up end to end. It assumes you’ve read Tool approval, which covers the full API. Here the focus is on the workflow structure: how to compose the gate with your agent, how the pause surfaces to an operator, and what happens on approve vs reject.

The pattern in one picture

agent calls send_email
       │
       ▼
engine emits TASK_AWAITING_APPROVAL
       │
       ▼
workflow pauses  ──► operator sees pending approval
                          │
              ┌───────────┴───────────┐
          approve                  reject
              │                       │
              ▼                       ▼
       task body runs;          ApprovalRejected raised;
       agent continues          agent sees error and adapts

The approval row is durable. If the worker restarts between the operator’s decision and the next checkpoint, the workflow replays and reads the recorded verdict from the row — the task body runs at most once.

Building the workflow

Step 1: define the tools

For this example, the agent has access to two tools: one that reads data and one that sends email. Only send_email needs approval.

from flux import ExecutionContext, task, workflow
from flux.tasks.ai import agent


@task
async def read_report(report_id: str) -> dict:
    """Read a report. Safe to run without approval."""
    return {"id": report_id, "summary": "Q3 revenue up 12%", "anomalies": []}


@task.with_options(requires_approval=True)
async def send_email(to: str, subject: str, body: str) -> dict:
    """Send an email. Requires human approval before running."""
    # In production: call your mail provider here
    return {"status": "sent", "to": to, "subject": subject}

requires_approval=True on send_email is the entire gating mechanism — there is no separate wrapper to apply.

Step 2: build the agent and workflow

@workflow
async def report_agent(ctx: ExecutionContext[dict]):
    raw = ctx.input or {}
    task_description = raw.get(
        "task",
        "Read the Q3 report and email a summary to the finance team.",
    )

    assistant = await agent(
        "You are a reporting assistant. Read reports and send summaries by email. "
        "Always read the report before composing the email.",
        model="ollama/llama3.2",
        name="report_assistant",
        tools=[read_report, send_email],
        stream=False,
    )

    return await assistant(task_description)

The agent loop runs normally until it tries to call send_email. At that point the engine emits TASK_AWAITING_APPROVAL, inserts an approval row keyed by the task call, and pauses the workflow. The execution stays in the paused state until an operator acts.

Running it

Register and kick off the workflow:

flux workflow register report_agent.py
flux workflow run report_agent '{"task": "Read report R-42 and email the summary to finance@example.com"}'

The command returns an execution ID. If the agent reaches send_email before anything else fails, the execution will be in the PAUSED state.

Find the pending approval:

flux execution approvals --execution <execution_id>

The row carries the task_call_id, the task’s arguments, the workflow + task name, and the timestamps. The operator can see exactly what the agent wants to send before deciding.

Approving or rejecting

Approve — the tool runs and the agent continues:

flux execution approve <execution_id> <task_call_id> --reason "lgtm"

Reject — the agent receives an error and adapts:

flux execution reject <execution_id> <task_call_id> --reason "not approved this quarter"

The same operations work over HTTP, useful when an external system (a ticket queue, a Slack bot, a custom dashboard) is the approver:

POST /executions/{execution_id}/approvals/{task_call_id}/approve
Content-Type: application/json

{"reason": "lgtm"}

A 409 response means another approver already decided this gate; the response carries current_status and decided_at without leaking the winning approver’s identity.

What happens after rejection

When the agent’s tool call is rejected, ApprovalRejected is raised inside the task call. The agent’s tool executor catches it and turns it into a normal tool-error response that the model sees:

Error: Approval rejected for task send_email by alice@oidc: not approved this quarter

The model treats it as a failed tool call and continues its loop. Depending on max_tool_calls, it may try a different approach, produce a final answer explaining what it could not do, or retry — in which case a new approval gate fires (each retry attempt is a fresh task call). The workflow body does not raise; rejection is a normal outcome in the agent’s context.

Replay safety

The approval row is the durable record. If the worker crashes after the operator’s decision but before the next checkpoint, Flux replays from the top: when it reaches the gated task, it reads the existing row, sees the recorded verdict, and either runs the body or re-raises ApprovalRejected without prompting again. The tool body runs at most once.

This is the same guarantee that covers any other paused workflow. See Pause and resume for the underlying replay model.

Variations

Multiple destructive tools with independent gates. Mark each task with requires_approval=True independently. Each pending call gets its own approval row and task_call_id; the operator decides them separately.

Conditional gating. Pass a callable to requires_approval so only some calls pause. The predicate receives the task’s arguments:

@task.with_options(
    requires_approval=lambda to, subject, body: not to.endswith("@example.com"),
)
async def send_email(to: str, subject: str, body: str) -> dict: ...

Emails to the internal domain go through unattended; everything else pauses.

Timeout the wait. There is no built-in timeout on an approval. If a deadline matters, set a task timeout:

@task.with_options(requires_approval=True, timeout=3600)
async def send_email(to: str, subject: str, body: str) -> dict: ...

If no one approves within an hour, the task call raises ExecutionTimeoutError and the workflow proceeds with normal error handling.

Autonomous mode for CI. Pass approval_mode="autonomous" to agent() to skip every approval gate inside that agent’s tool batch:

assistant = await agent(
    "...",
    model="ollama/llama3.2",
    tools=[read_report, send_email],
    approval_mode="autonomous",
)

Conditional approval in an external service. When an external approver decides, fetch the pending row via GET /executions/{id}/approvals/{task_call_id} (or flux execution approvals --execution <id> --json), inspect the arguments, and post the verdict back.

What’s next