Approval Gates: Pausing Agent Workflows for Human Sign-Off

Autonomous agents are powerful, but some actions need a human in the loop. Sending an email, deleting a database record, making a payment — these aren’t decisions you want an agent making alone. Approval gates give you controlled autonomy: the agent does the thinking, but humans approve the doing.

Why Approval Gates

The core tension in agent design is between autonomy and safety. Too much autonomy and agents make costly mistakes. Too little and they’re just expensive form-filling tools.

Approval gates find the middle ground:

• Agent proposes — The agent plans the action, gathers context, and presents a recommendation
• Human disposes — A human reviews, approves, rejects, or modifies the action
• Agent executes — Only approved actions are carried out

This pattern is essential for:

• Financial transactions (payments, transfers, refunds)
• External communications (emails, messages, social posts)
• Data mutations (deletes, overwrites, schema changes)
• Compliance-sensitive actions (legal, medical, regulatory)

Implementation Patterns

LangGraph with Interrupt

LangGraph’s interrupt function pauses execution and waits for human input:

from langgraph.graph import StateGraph, MessagesState, START, END
from langgraph.types import interrupt
from langgraph.checkpoint.memory import MemorySaver

def draft_email(state: MessagesState):
    """Agent drafts an email for review."""
    last_message = state["messages"][-1].content
    
    draft = llm.invoke(f"""Draft a professional email based on this request:
    
{last_message}

Include subject line, greeting, body, and signature.""")
    
    return {
        "messages": [draft],
        "pending_action": {
            "type": "send_email",
            "content": draft.content,
            "recipient": extract_recipient(last_message),
        }
    }

def approval_gate(state: MessagesState):
    """Pause and wait for human approval."""
    pending = state.get("pending_action", {})
    
    # This pauses execution and shows the pending action to the human
    human_decision = interrupt({
        "action_type": pending.get("type"),
        "content": pending.get("content"),
        "instructions": "Approve, reject, or modify this action"
    })
    
    return {"approval_decision": human_decision}

def execute_action(state: MessagesState):
    """Execute only if approved."""
    decision = state.get("approval_decision", {})
    
    if decision.get("approved"):
        # Execute the action
        if state["pending_action"]["type"] == "send_email":
            send_email(
                to=state["pending_action"]["recipient"],
                subject=decision.get("subject", "No Subject"),
                body=decision.get("content", state["pending_action"]["content"])
            )
        return {"messages": [("ai", "Action executed successfully")]}
    else:
        return {"messages": [("ai", f"Action rejected: {decision.get('reason', 'No reason provided')}")]}

# Build the graph
graph = StateGraph(MessagesState)
graph.add_node("draft", draft_email)
graph.add_node("approve", approval_gate)
graph.add_node("execute", execute_action)

graph.add_edge(START, "draft")
graph.add_edge("draft", "approve")
graph.add_conditional_edges(
    "approve",
    lambda s: "execute" if s.get("approval_decision", {}).get("approved") else END
)
graph.add_edge("execute", END)

checkpointer = MemorySaver()
app = graph.compile(checkpointer=checkpointer)

Reacting to Approval

The human provides their decision via update_state:

config = {"configurable": {"thread_id": "email-draft-1"}}

# Agent runs until approval gate
result = app.invoke({"messages": [("user", "Send a follow-up email to john@example.com")]}, config)

# Human reviews and provides decision
app.update_state(
    config,
    {
        "approval_decision": {
            "approved": True,
            "subject": "Following Up on Our Discussion",
            "modifications": ["Make the tone more casual"]
        }
    }
)

# Resume execution
result = app.invoke(None, config)

Conditional Approval Gates

Not every action needs the same level of scrutiny:

class ApprovalConfig:
    def __init__(self):
        self.gates = {
            "read_data": {"required": False},
            "write_data": {"required": True, "timeout_seconds": 300},
            "send_email": {"required": True, "timeout_seconds": 3600},
            "delete_record": {"required": True, "approver_role": "admin"},
            "payment": {"required": True, "approver_role": "finance", "timeout_seconds": 86400},
        }
    
    def needs_approval(self, action_type: str) -> bool:
        gate = self.gates.get(action_type, {"required": True})
        return gate["required"]
    
    def get_timeout(self, action_type: str) -> int:
        gate = self.gates.get(action_type, {"timeout_seconds": 300})
        return gate["timeout_seconds"]

UI for Approval Workflows

Web Dashboard

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel

app = FastAPI()

class ApprovalDecision(BaseModel):
    approved: bool
    reason: str = ""
    modifications: dict = {}

@app.get("/pending-approvals/{session_id}")
async def get_pending(session_id: str):
    """Get pending approval requests for a session."""
    pending = await get_pending_approvals(session_id)
    return {"approvals": pending}

@app.post("/approve/{approval_id}")
async def approve_action(approval_id: str, decision: ApprovalDecision):
    """Submit an approval decision."""
    await submit_decision(approval_id, decision)
    return {"status": "decision_recorded"}

@app.get("/approval-status/{approval_id}")
async def check_status(approval_id: str):
    """Check if an approval has been made."""
    status = await get_approval_status(approval_id)
    return status

Slack Integration

import slack_sdk

def send_approval_to_slack(approval_request: dict) -> str:
    """Send approval request to Slack and wait for response."""
    client = slack_sdk.WebClient(token=os.environ["SLACK_BOT_TOKEN"])
    
    # Send message with buttons
    response = client.chat_postMessage(
        channel="#agent-approvals",
        text=f"Action requires approval: {approval_request['action_type']}",
        blocks=[
            {
                "type": "section",
                "text": {
                    "type": "mrkdwn",
                    "text": f"*Agent Action Request*\n\n"
                            f"*Type:* {approval_request['action_type']}\n"
                            f"*Details:* {approval_request['content'][:200]}"
                }
            },
            {
                "type": "actions",
                "elements": [
                    {
                        "type": "button",
                        "text": {"type": "plain_text", "text": "Approve"},
                        "action_id": f"approve_{approval_request['id']}",
                        "style": "primary"
                    },
                    {
                        "type": "button",
                        "text": {"type": "plain_text", "text": "Reject"},
                        "action_id": f"reject_{approval_request['id']}",
                        "style": "danger"
                    }
                ]
            }
        ]
    )
    
    return response["ts"]

Handling Timeouts

Agents shouldn’t block forever waiting for approval:

import asyncio
from datetime import datetime, timedelta

class ApprovalTimeoutHandler:
    def __init__(self, default_timeout: int = 300):
        self.default_timeout = default_timeout
    
    async def wait_for_approval(
        self,
        approval_id: str,
        timeout: int = None
    ) -> dict:
        timeout = timeout or self.default_timeout
        start = datetime.now()
        
        while (datetime.now() - start).seconds < timeout:
            status = await get_approval_status(approval_id)
            
            if status["decided"]:
                return status["decision"]
            
            await asyncio.sleep(5)  # Poll every 5 seconds
        
        # Timeout reached
        return {
            "approved": False,
            "reason": "Approval timeout",
            "timed_out": True
        }

Best Practices

Clear Context

Always provide enough context for informed decisions:

def build_approval_context(action: dict, agent_state: dict) -> dict:
    return {
        "action_type": action["type"],
        "action_details": action["content"],
        "reasoning": agent_state.get("reasoning", "No reasoning provided"),
        "risk_level": assess_risk(action),
        "similar_past_actions": get_similar_actions(action["type"]),
        "estimated_impact": estimate_impact(action),
    }

Audit Trail

Log every approval decision:

class ApprovalAuditLog:
    def log_decision(
        self,
        approval_id: str,
        action: dict,
        decision: dict,
        reviewer: str
    ):
        log_entry = {
            "timestamp": datetime.now().isoformat(),
            "approval_id": approval_id,
            "action_type": action["type"],
            "action_content_hash": hash(action["content"]),
            "approved": decision["approved"],
            "reviewer": reviewer,
            "reason": decision.get("reason", ""),
            "response_time_ms": calculate_response_time(approval_id),
        }
        
        self._store(log_entry)

Escalation

If primary approvers don’t respond, escalate:

class EscalationPolicy:
    def __init__(self):
        self.escalation_chain = [
            {"level": 1, "timeout": 300, "notify": ["team-lead"]},
            {"level": 2, "timeout": 600, "notify": ["engineering-manager"]},
            {"level": 3, "timeout": 1800, "notify": ["cto"]},
        ]
    
    def get_next_level(self, current_level: int) -> dict:
        if current_level < len(self.escalation_chain):
            return self.escalation_chain[current_level]
        return None

Conclusion

Approval gates are the bridge between autonomous agents and responsible systems. They let agents do the thinking while humans control the doing. Start with gates on high-risk actions, add clear context and audit trails, handle timeouts gracefully, and build escalation policies for when approvers are unavailable.