---
title: "Building an Approval Workflow Agent: Human-in-the-Loop for Sensitive Actions"
description: "Build an AI agent with human approval gates for sensitive actions, including notification systems, configurable timeout handling, delegation chains, and audit logging for compliance."
canonical: https://callsphere.ai/blog/building-approval-workflow-agent-human-in-the-loop-sensitive-actions
category: "Learn Agentic AI"
tags: ["Human-in-the-Loop", "Approval Workflow", "Agent Safety", "Agentic AI", "Python"]
author: "CallSphere Team"
published: 2026-03-17T00:00:00.000Z
updated: 2026-05-07T06:52:42.255Z
---

# Building an Approval Workflow Agent: Human-in-the-Loop for Sensitive Actions

> Build an AI agent with human approval gates for sensitive actions, including notification systems, configurable timeout handling, delegation chains, and audit logging for compliance.

## Why Agents Need Approval Gates

Autonomous AI agents are powerful, but certain actions carry real-world consequences that demand human oversight. Sending an email to a customer, executing a financial transaction, deploying code to production, or deleting records — these actions should not happen without explicit human approval.

An **approval workflow** inserts a pause point between the agent deciding to act and the action actually executing. The agent proposes the action, a human reviews it, and only after approval does execution proceed.

## Designing the Approval System

The core abstraction is an approval request that captures the proposed action, who requested it, who can approve it, and the current status:

```mermaid
flowchart LR
    INPUT(["User intent"])
    PARSE["Parse plus
classify"]
    PLAN["Plan and tool
selection"]
    AGENT["Agent loop
LLM plus tools"]
    GUARD{"Guardrails
and policy"}
    EXEC["Execute and
verify result"]
    OBS[("Trace and metrics")]
    OUT(["Outcome plus
next action"])
    INPUT --> PARSE --> PLAN --> AGENT --> GUARD
    GUARD -->|Pass| EXEC --> OUT
    GUARD -->|Fail| AGENT
    AGENT --> OBS
    style AGENT fill:#4f46e5,stroke:#4338ca,color:#fff
    style GUARD fill:#f59e0b,stroke:#d97706,color:#1f2937
    style OBS fill:#ede9fe,stroke:#7c3aed,color:#1e1b4b
    style OUT fill:#059669,stroke:#047857,color:#fff
```

```python
import uuid
import asyncio
from datetime import datetime, timezone, timedelta
from dataclasses import dataclass, field
from enum import Enum
from typing import Any

class ApprovalStatus(Enum):
    PENDING = "pending"
    APPROVED = "approved"
    REJECTED = "rejected"
    TIMED_OUT = "timed_out"
    DELEGATED = "delegated"

@dataclass
class ApprovalRequest:
    request_id: str = field(default_factory=lambda: str(uuid.uuid4()))
    action_type: str = ""
    action_params: dict[str, Any] = field(default_factory=dict)
    requested_by: str = ""
    approvers: list[str] = field(default_factory=list)
    status: ApprovalStatus = ApprovalStatus.PENDING
    created_at: datetime = field(default_factory=lambda: datetime.now(timezone.utc))
    timeout_minutes: int = 30
    decision_by: str = ""
    decision_reason: str = ""
    decided_at: datetime | None = None
```

## The Approval Manager

The approval manager handles the lifecycle of requests — creating them, waiting for decisions, and enforcing timeouts:

```python
class ApprovalManager:
    def __init__(self):
        self.requests: dict[str, ApprovalRequest] = {}
        self._waiters: dict[str, asyncio.Event] = {}

    async def request_approval(
        self,
        action_type: str,
        action_params: dict,
        approvers: list[str],
        timeout_minutes: int = 30,
    ) -> ApprovalRequest:
        """Create an approval request and wait for a decision."""
        req = ApprovalRequest(
            action_type=action_type,
            action_params=action_params,
            approvers=approvers,
            timeout_minutes=timeout_minutes,
        )
        self.requests[req.request_id] = req
        self._waiters[req.request_id] = asyncio.Event()

        # Notify approvers
        await self._notify_approvers(req)

        # Wait for decision or timeout
        try:
            await asyncio.wait_for(
                self._waiters[req.request_id].wait(),
                timeout=timeout_minutes * 60,
            )
        except asyncio.TimeoutError:
            req.status = ApprovalStatus.TIMED_OUT

        return req

    def submit_decision(
        self,
        request_id: str,
        approved: bool,
        decided_by: str,
        reason: str = "",
    ):
        """An approver submits their decision."""
        req = self.requests.get(request_id)
        if not req or req.status != ApprovalStatus.PENDING:
            raise ValueError("Invalid or already decided request")

        if decided_by not in req.approvers:
            raise PermissionError(f"{decided_by} is not an authorized approver")

        req.status = ApprovalStatus.APPROVED if approved else ApprovalStatus.REJECTED
        req.decision_by = decided_by
        req.decision_reason = reason
        req.decided_at = datetime.now(timezone.utc)

        # Wake up the waiting coroutine
        self._waiters[request_id].set()

    async def _notify_approvers(self, req: ApprovalRequest):
        """Send notifications to all potential approvers."""
        for approver in req.approvers:
            await send_notification(
                to=approver,
                subject=f"Approval needed: {req.action_type}",
                body=f"Action: {req.action_type}\nParams: {req.action_params}",
            )
```

The key design choice is using `asyncio.Event` and `asyncio.wait_for`. The agent's coroutine suspends without consuming resources while waiting for a human decision. If the timeout expires, the request is automatically marked as timed out.

## Integrating with an Agent

Wrap your agent's sensitive actions with approval checks:

```python
class ApprovalGatedAgent:
    SENSITIVE_ACTIONS = {"send_email", "delete_record", "execute_payment"}

    def __init__(self, approval_manager: ApprovalManager):
        self.approvals = approval_manager

    async def execute_action(self, action: str, params: dict) -> dict:
        if action not in self.SENSITIVE_ACTIONS:
            return await self._run_action(action, params)

        # Request approval for sensitive actions
        req = await self.approvals.request_approval(
            action_type=action,
            action_params=params,
            approvers=["manager@company.com", "lead@company.com"],
            timeout_minutes=60,
        )

        if req.status == ApprovalStatus.APPROVED:
            result = await self._run_action(action, params)
            return {"status": "executed", "approved_by": req.decision_by, **result}
        elif req.status == ApprovalStatus.REJECTED:
            return {"status": "rejected", "reason": req.decision_reason}
        else:
            return {"status": "timed_out", "message": "No approver responded"}

    async def _run_action(self, action: str, params: dict) -> dict:
        # Dispatch to actual implementation
        handler = getattr(self, f"_action_{action}", None)
        if handler:
            return await handler(params)
        return {"error": f"Unknown action: {action}"}
```

## Delegation Chains

Sometimes the initial approver is unavailable. Delegation allows an approver to forward the request:

```python
def delegate_approval(
    self,
    request_id: str,
    delegated_by: str,
    delegate_to: str,
):
    """Delegate an approval request to another person."""
    req = self.requests.get(request_id)
    if not req or req.status != ApprovalStatus.PENDING:
        raise ValueError("Cannot delegate a non-pending request")

    if delegated_by not in req.approvers:
        raise PermissionError("Not an authorized approver")

    # Add the delegate and record the chain
    req.approvers.append(delegate_to)
    req.action_params.setdefault("delegation_chain", [])
    req.action_params["delegation_chain"].append({
        "from": delegated_by,
        "to": delegate_to,
        "at": datetime.now(timezone.utc).isoformat(),
    })
```

## Building the Notification Layer

Notifications can use any channel — email, Slack, or a web dashboard. Here is a simple multi-channel notifier:

```python
async def send_notification(to: str, subject: str, body: str):
    """Route notifications based on recipient preferences."""
    channel = get_preferred_channel(to)

    if channel == "slack":
        await post_slack_message(to, f"*{subject}*\n{body}")
    elif channel == "email":
        await send_email(to, subject, body)
    elif channel == "webhook":
        await post_webhook(to, {"subject": subject, "body": body})
```

## FAQ

### How do I prevent the agent from bypassing the approval gate?

Enforce approval at the execution layer, not the agent layer. The tool implementations themselves should check for valid approval tokens before executing. Even if the agent skips the approval call, the underlying `send_email` or `execute_payment` function refuses to run without a signed approval reference.

### What timeout value should I use for approval requests?

It depends on the action's urgency. For customer-facing emails, 30 to 60 minutes is reasonable. For financial transactions, you might want 15 minutes with escalation. For non-urgent batch operations, 24 hours with a daily digest notification works well. Always provide a fallback behavior — either auto-reject on timeout or escalate to a backup approver.

### How do I handle approvals when the agent runs outside business hours?

Implement an escalation policy with timezone awareness. If the primary approver is outside business hours, route to an on-call approver or queue the request with a "next business day" timeout. For critical actions, maintain a rotating on-call schedule of approvers who receive notifications regardless of time.

---

#HumanintheLoop #ApprovalWorkflow #AgentSafety #AgenticAI #Python #LearnAI #AIEngineering

---

Source: https://callsphere.ai/blog/building-approval-workflow-agent-human-in-the-loop-sensitive-actions