Human-in-the-Loop Approval Workflows

Autonomous agents can take irreversible real-world actions: send emails, delete database records, make payments, execute code, post to social media, grant permissions. Unlike a model that produces text, an agent that calls a tool has side effects in the world. A poorly-aligned response can be corrected in the next turn. A sent wire transfer cannot. A human-in-the-loop (HITL) gate is the last line of defence between an AI agent and an irreversible action. This reference architecture describes how to build one using Cognisafe’s webhook system and your existing approval tooling.

Two variants

This pattern has two variants with different trade-offs:

	Variant 1: Pre-execution gate	Variant 2: Post-detection review
When	Before the dangerous action executes	After the dangerous action completes
Effect	Prevents the action if rejected	Creates an audit trail; informs future policy
Latency	Adds human response time (minutes to hours)	Zero added latency
Framework requirement	Agent must support async tool execution	None — works with any agent
Use for	Irreversible actions (payments, deletes, emails)	Monitoring and policy calibration

For truly irreversible actions, use Variant 1. For actions that can be unwound or that are lower stakes, Variant 2 is sufficient and imposes no user-visible latency.

Variant 1: Pre-execution gate

How it works

Agent decides to call a dangerous tool
          │
          ▼
Approval-gating wrapper intercepts the tool call
          │
          ▼
Submit approval request to approval service
(ServiceNow workflow / Jira approval / Slack bot / custom FastAPI)
          │
          ▼
Approval service notifies human approver
(email, Slack, PagerDuty, mobile app)
          │
          ├── Approved ──▶  Execute the real tool · return result to agent
          │
          └── Rejected ──▶  Raise PermissionError · agent handles gracefully
                            (or: timeout ──▶ fail-safe reject)

Cognisafe logs the tool call in both paths:
  - Audit trail of what was requested, who approved/rejected, when
  - Fires threat_detected webhook if the tool name matches high-risk classifier

Framework requirements

Variant 1 requires the agent framework to support intercepting tool calls before execution. This works with:

LangGraph — add a conditional edge before the tool node that routes to an approval subgraph
Semantic Kernel — implement a FunctionInvocationFilter middleware
Custom MCP server — wrap dangerous tool handlers in the approval gate
LangChain custom tools — override _run / _arun with the gate wrapper

It does not work with frameworks where tool execution is entirely managed by the LLM provider (e.g., OpenAI Assistants with hosted tools). In those cases, use Variant 2 for monitoring and redesign the agent to use a self-hosted tool server.

Approval-gating wrapper

import hashlib, hmac, time
import httpx

# Configuration
APPROVAL_SERVICE_URL = "https://approvals.yourco.com"
APPROVAL_SERVICE_TOKEN = "your-approval-service-token"
COGNISAFE_API_URL = "https://api.cognisafe.io"
COGNISAFE_API_KEY = "your-cognisafe-api-key"

# High-risk tools that always require approval
HIGH_RISK_TOOLS = {
    # Financial
    "transfer_funds", "create_payment", "refund_order", "update_billing",
    # Destructive
    "delete_record", "drop_table", "terminate_instance", "purge_queue",
    # External communication
    "send_email", "post_to_social", "create_pr", "merge_pr", "send_sms",
    # Privilege
    "add_user", "grant_role", "update_permissions", "delete_user",
}

def gated_tool(tool_name: str, args: dict, agent_name: str = "unknown") -> dict:
    """
    Wrap any dangerous tool call in a HITL approval gate.
    Blocks until approved, rejected, or timed out.
    """
    if tool_name not in HIGH_RISK_TOOLS:
        # Not a high-risk tool — execute directly
        return execute_tool(tool_name, args)

    # Get the current Cognisafe request ID for audit trail linkage
    cognisafe_request_id = get_current_cognisafe_request_id()

    # 1. Submit approval request
    resp = httpx.post(
        f"{APPROVAL_SERVICE_URL}/requests",
        json={
            "tool": tool_name,
            "args": args,
            "agent": agent_name,
            "cognisafe_request_id": cognisafe_request_id,
            "requested_at": time.time(),
            "context": {
                "why": "Agent requested this action during task execution",
                "risk": describe_risk(tool_name),
                "owasp": "LLM06 - Excessive Agency",
                "review_url": f"{COGNISAFE_API_URL}/requests/{cognisafe_request_id}",
            }
        },
        headers={"Authorization": f"Bearer {APPROVAL_SERVICE_TOKEN}"},
        timeout=10,
    )
    resp.raise_for_status()
    ticket_id = resp.json()["ticket_id"]

    # 2. Poll for decision (max 5 minutes, poll every 10 seconds)
    max_polls = 30
    for attempt in range(max_polls):
        time.sleep(10)
        status_resp = httpx.get(
            f"{APPROVAL_SERVICE_URL}/requests/{ticket_id}",
            headers={"Authorization": f"Bearer {APPROVAL_SERVICE_TOKEN}"},
            timeout=5,
        )
        status = status_resp.json()

        if status["decision"] == "approved":
            # Log approval in Cognisafe audit trail
            log_approval_decision(cognisafe_request_id, ticket_id, "approved", status)
            return execute_tool(tool_name, args)

        elif status["decision"] == "rejected":
            log_approval_decision(cognisafe_request_id, ticket_id, "rejected", status)
            raise PermissionError(
                f"Action '{tool_name}' was rejected by {status.get('approver', 'an approver')}. "
                f"Reason: {status.get('reason', 'No reason provided.')} "
                f"Ticket: {ticket_id}"
            )

    # Timeout: fail-safe reject (never auto-approve on timeout)
    log_approval_decision(cognisafe_request_id, ticket_id, "timeout", {})
    raise TimeoutError(
        f"Approval for '{tool_name}' timed out after {max_polls * 10} seconds. "
        f"Action has been blocked. Ticket: {ticket_id}"
    )


def describe_risk(tool_name: str) -> str:
    risk_descriptions = {
        "transfer_funds":      "Irreversible financial transfer. Cannot be unwound without manual intervention.",
        "delete_record":       "Permanent data deletion. Recoverable only from backup.",
        "send_email":          "External communication sent to real recipients.",
        "grant_role":          "Privilege escalation. Could give a user unauthorised access.",
        "terminate_instance":  "Infrastructure destruction. May cause service outage.",
        "merge_pr":            "Code change merged to production branch.",
    }
    return risk_descriptions.get(tool_name, "High-risk action — review before approving.")


def log_approval_decision(request_id: str, ticket_id: str, decision: str, status: dict) -> None:
    """Post the approval decision back to Cognisafe for the audit trail."""
    try:
        httpx.post(
            f"{COGNISAFE_API_URL}/api/requests/{request_id}/annotations",
            json={
                "type": "hitl_decision",
                "ticket_id": ticket_id,
                "decision": decision,
                "approver": status.get("approver"),
                "reason": status.get("reason"),
            },
            headers={"Authorization": f"Bearer {COGNISAFE_API_KEY}"},
            timeout=5,
        )
    except Exception:
        pass  # Non-blocking — audit log failure must not block the tool gate

Never auto-approve on timeout. The fail-safe must be rejection. An agent that can time out into execution has no gate — it just has a delay. If the approval SLA cannot be met, consider whether the action is suitable for automation at all.

LangGraph integration

from langgraph.graph import StateGraph, END
from langgraph.prebuilt import ToolNode
from typing import TypedDict, Annotated
import operator

class AgentState(TypedDict):
    messages: Annotated[list, operator.add]
    pending_tool_call: dict | None
    approval_status: str | None

def should_gate(state: AgentState) -> str:
    """Route to approval gate if the pending tool call is high-risk."""
    pending = state.get("pending_tool_call")
    if pending and pending["name"] in HIGH_RISK_TOOLS:
        return "approval_gate"
    return "execute_tool"

def approval_gate_node(state: AgentState) -> AgentState:
    """Block and wait for human approval."""
    pending = state["pending_tool_call"]
    try:
        result = gated_tool(
            tool_name=pending["name"],
            args=pending["args"],
            agent_name="my-agent",
        )
        return {**state, "approval_status": "approved", "pending_tool_call": None}
    except PermissionError as e:
        return {**state, "approval_status": f"rejected: {e}", "pending_tool_call": None}
    except TimeoutError as e:
        return {**state, "approval_status": f"timeout: {e}", "pending_tool_call": None}

# Build the graph
graph = StateGraph(AgentState)
graph.add_node("agent", agent_node)
graph.add_node("approval_gate", approval_gate_node)
graph.add_node("tool_executor", ToolNode(safe_tools))

graph.add_conditional_edges("agent", should_gate, {
    "approval_gate": "approval_gate",
    "execute_tool": "tool_executor",
})
graph.add_edge("approval_gate", "agent")  # Return to agent after decision
graph.add_edge("tool_executor", "agent")

Slack approval bot

The simplest approval service for many teams is a Slack bot with Approve/Reject buttons. This avoids deploying a separate service and integrates with where engineers already work.

# approval_bot.py — FastAPI-based Slack approval bot
from fastapi import FastAPI, Request
from slack_sdk import WebClient
from slack_sdk.errors import SlackApiError
import json, uuid

app = FastAPI()
slack = WebClient(token="xoxb-your-slack-bot-token")

# In-memory store (use Redis in production for multi-instance deployments)
pending_approvals: dict[str, dict] = {}

APPROVAL_CHANNEL = "#ai-hitl-approvals"

@app.post("/requests")
async def create_approval_request(request: Request) -> dict:
    body = await request.json()
    ticket_id = str(uuid.uuid4())

    pending_approvals[ticket_id] = {
        "ticket_id": ticket_id,
        "decision": "pending",
        "approver": None,
        "reason": None,
        **body,
    }

    # Build Slack Block Kit message
    blocks = [
        {
            "type": "header",
            "text": {"type": "plain_text", "text": ":warning: Agent Action Requires Approval"}
        },
        {
            "type": "section",
            "fields": [
                {"type": "mrkdwn", "text": f"*Tool:* `{body['tool']}`"},
                {"type": "mrkdwn", "text": f"*Agent:* `{body['agent']}`"},
                {"type": "mrkdwn", "text": f"*Risk:* {body.get('context', {}).get('risk', 'Unknown')}"},
                {"type": "mrkdwn", "text": f"*OWASP:* {body.get('context', {}).get('owasp', 'LLM06')}"},
            ]
        },
        {
            "type": "section",
            "text": {"type": "mrkdwn", "text": f"*Arguments:*\n```{json.dumps(body['args'], indent=2)[:500]}```"}
        },
        {
            "type": "section",
            "text": {"type": "mrkdwn", "text": f"<{body.get('context', {}).get('review_url', '')}|View full request in Cognisafe>"}
        },
        {
            "type": "actions",
            "elements": [
                {
                    "type": "button",
                    "text": {"type": "plain_text", "text": "Approve"},
                    "style": "primary",
                    "value": ticket_id,
                    "action_id": "approve_action",
                },
                {
                    "type": "button",
                    "text": {"type": "plain_text", "text": "Reject"},
                    "style": "danger",
                    "value": ticket_id,
                    "action_id": "reject_action",
                }
            ]
        }
    ]

    slack.chat_postMessage(channel=APPROVAL_CHANNEL, blocks=blocks)
    return {"ticket_id": ticket_id, "status": "pending"}


@app.get("/requests/{ticket_id}")
async def get_approval_status(ticket_id: str) -> dict:
    approval = pending_approvals.get(ticket_id)
    if not approval:
        return {"ticket_id": ticket_id, "decision": "not_found"}
    return approval


@app.post("/slack/interactions")
async def handle_slack_interaction(request: Request) -> dict:
    form = await request.form()
    payload = json.loads(form["payload"])

    action = payload["actions"][0]
    ticket_id = action["value"]
    action_id = action["action_id"]
    approver = payload["user"]["name"]

    if ticket_id not in pending_approvals:
        return {"text": "Approval request not found or already resolved."}

    if action_id == "approve_action":
        pending_approvals[ticket_id]["decision"] = "approved"
        pending_approvals[ticket_id]["approver"] = approver
        message = f":white_check_mark: Approved by {approver}"
    else:
        pending_approvals[ticket_id]["decision"] = "rejected"
        pending_approvals[ticket_id]["approver"] = approver
        message = f":x: Rejected by {approver}"

    # Update the Slack message to show the decision
    slack.chat_update(
        channel=payload["channel"]["id"],
        ts=payload["message"]["ts"],
        text=message,
        blocks=[{
            "type": "section",
            "text": {"type": "mrkdwn", "text": f"{message}\nTool: `{pending_approvals[ticket_id]['tool']}`"}
        }]
    )

    return {"status": "ok"}

Variant 2: Post-detection review

When pre-execution gating is not feasible, configure Cognisafe to detect and alert on high-risk tool calls after they complete. This creates an audit trail and informs future policy even if it does not block the action.

Configure the custom scorer

# evals/scorers.yaml
scorers:
  - name: dangerous_tool_detector
    type: keyword_list
    severity: critical
    description: "Detects calls to high-risk tools in agent output"
    keywords:
      # Financial
      - "transfer_funds"
      - "create_payment"
      - "refund_order"
      # Destructive
      - "delete_record"
      - "drop_table"
      - "terminate_instance"
      # External communication
      - "send_email"
      - "post_to_social"
      - "create_pr"
      # Privilege
      - "add_user"
      - "grant_role"
      - "update_permissions"
    # Fire threat_detected webhook on any match
    webhook_on_match: true

Webhook receiver: post-detection review ticket

async def handle_post_detection(event: dict) -> None:
    """
    Create a review ticket for a completed high-risk action.
    This is audit trail + policy feedback, not a gate.
    """
    data = event["data"]

    # Create a Jira review ticket
    await create_jira_ticket(event, {
        "summary": f"[HITL Review] {data['agent_name']} executed high-risk tool",
        "description": (
            f"A high-risk tool call was detected in a completed agent interaction.\n\n"
            f"This ticket is for review and policy purposes — the action has already executed.\n\n"
            f"Tool pattern detected: {data.get('prompt_snippet', 'see report')}\n"
            f"Agent: {data['agent_name']} ({data.get('agent_tag', 'untagged')})\n"
            f"Full interaction: {data.get('report_url', 'N/A')}\n\n"
            f"Action: Review the interaction. If the action was inappropriate, "
            f"add a pre-execution gate for this tool using Variant 1 of the HITL pattern.\n"
            f"Docs: https://docs.cognisafe.io/reference-architectures/hitl-approval"
        ),
        "labels": ["hitl-review", "ai-security", "owasp-llm06"],
    })

High-risk tool classification

Not all tools need HITL. Apply the gate selectively to preserve agent velocity for low-risk operations. Classify tools by the reversibility of their side effects and the blast radius of a mistake.

Category	Tools	HITL required	Rationale
Financial	`transfer_funds`, `create_payment`, `refund_order`, `update_billing`	Always	Irreversible without manual intervention; regulatory exposure
Destructive	`delete_record`, `drop_table`, `terminate_instance`, `purge_queue`	Always	Data or infrastructure loss; recovery from backup only
External communication	`send_email`, `post_to_social`, `create_pr`, `merge_pr`, `send_sms`	Always	Reputational risk; cannot un-send
Privilege escalation	`add_user`, `grant_role`, `update_permissions`, `delete_user`	Always	Security boundary violation; enables lateral movement
External API calls	`call_external_api`, `submit_form`, `place_order`	Depends	Gate if the API has side effects; safe if read-only
Read-only operations	`search_db`, `list_files`, `get_record`, `fetch_url`	Never	No side effects; no gating needed

Start with a conservative list and expand it. It is better to gate too many tools initially and relax the list based on analyst feedback than to discover after an incident that a tool was unguarded. Review the high-risk tool list quarterly with the AI system owners and the security team.

Custom scorer for dangerous tool detection

# evals/scorers.yaml — production configuration
scorers:
  - name: hitl_tool_detector
    type: keyword_list
    severity: critical
    description: "Detects high-risk tool names in agent interactions"
    # Matches against both the request body (tool call) and response body (tool result)
    keywords:
      - transfer_funds
      - create_payment
      - refund_order
      - delete_record
      - drop_table
      - terminate_instance
      - purge_queue
      - send_email
      - post_to_social
      - create_pr
      - merge_pr
      - add_user
      - grant_role
      - update_permissions
      - delete_user
    webhook_on_match: true
    # Exclude red team runs from creating P1 alerts
    exclude_tags:
      - redteam
      - staging
      - test

Compliance value

OWASP LLM06 — Excessive Agency

HITL is the primary control for LLM06. The OWASP guidance states: “Limit the permissions of LLM-integrated systems and implement human-in-the-loop controls for actions that are irreversible or have significant impact.” Document the approval gate as the LLM06 control in your security posture assessment.

EU AI Act Article 14 — Human Oversight

For AI systems classified as “high-risk” under the EU AI Act (Annex III), Article 14 requires that natural persons are able to “oversee the functioning” of the AI system and “intervene or interrupt” it. A pre-execution HITL gate satisfies this requirement for the specific actions it covers. Document each tool category in your AI Act technical documentation.

SOC 2 CC6.1 — Logical Access Controls

The approval service functions as a logical access control: the AI agent cannot execute a privileged action without an authorised human granting access for that specific invocation. Include the HITL gate in your CC6.1 control evidence:

“High-risk tool calls made by AI agents require approval from an authorised human operator before execution. Approvals are logged in the Cognisafe audit trail with the approver identity, timestamp, and decision. The approval gate is implemented as a synchronous block in the agent tool wrapper and defaults to rejection on timeout.”

Operational considerations

Latency

HITL adds human response time. An approval workflow that requires a human to click Approve adds seconds to minutes to hours of latency to the agent task. Design agent tasks accordingly:

Set agent timeout longer than your approval SLA: if your Slack approval SLA is 5 minutes, set the agent task timeout to 10 minutes
Inform end users that high-risk actions are queued for human review: “I need to transfer funds — this action is pending approval and will complete within 5 minutes”
Define a business-hours policy: if approvals are only reviewed during working hours, agents should not initiate high-risk actions outside those hours unless there is an on-call roster

Approval fatigue

If the approval queue receives too many requests, approvers start rubber-stamping without reviewing. Counter this by:

Calibrating the high-risk tool list — do not gate low-stakes operations
Providing maximum context in the approval message so the review takes 30 seconds, not 5 minutes
Setting a volume alert: if more than 10 approvals are pending simultaneously, something is wrong with the agent

Failure mode: approval service unavailable

If the approval service is unavailable, the gate wrapper will raise an exception when it cannot submit the request. The agent will receive a tool error and should surface this to the user. This is the correct behaviour — unavailability of the approval service must not result in ungated execution. Design the gate wrapper to fail closed.

def gated_tool(tool_name: str, args: dict, agent_name: str = "unknown") -> dict:
    if tool_name not in HIGH_RISK_TOOLS:
        return execute_tool(tool_name, args)

    try:
        resp = httpx.post(f"{APPROVAL_SERVICE_URL}/requests", ...)
        resp.raise_for_status()
    except (httpx.ConnectError, httpx.TimeoutException, httpx.HTTPStatusError) as e:
        # Approval service unavailable — fail closed, never execute
        raise RuntimeError(
            f"Approval service unavailable for '{tool_name}'. "
            f"Action blocked. Contact the platform team. Error: {e}"
        ) from e

Getting Started

SDKs

LLM Providers

Self-hosted Models

Self-hosting

Integrations

Reference Architectures

Deployment Blueprints

Safety & Scoring

Human-in-the-Loop Approval Workflows

Two variants

Variant 1: Pre-execution gate

How it works

Framework requirements

Approval-gating wrapper

LangGraph integration

Slack approval bot

Variant 2: Post-detection review

Configure the custom scorer

Webhook receiver: post-detection review ticket

High-risk tool classification

Custom scorer for dangerous tool detection

Compliance value

OWASP LLM06 — Excessive Agency

EU AI Act Article 14 — Human Oversight

SOC 2 CC6.1 — Logical Access Controls

Operational considerations

Latency

Approval fatigue

Failure mode: approval service unavailable

​Two variants

​Variant 1: Pre-execution gate

​How it works

​Framework requirements

​Approval-gating wrapper

​LangGraph integration

​Slack approval bot

​Variant 2: Post-detection review

​Configure the custom scorer

​Webhook receiver: post-detection review ticket

​High-risk tool classification

​Custom scorer for dangerous tool detection

​Compliance value

​OWASP LLM06 — Excessive Agency

​EU AI Act Article 14 — Human Oversight

​SOC 2 CC6.1 — Logical Access Controls

​Operational considerations

​Latency

​Approval fatigue

​Failure mode: approval service unavailable

Two variants

Variant 1: Pre-execution gate

How it works

Framework requirements

Approval-gating wrapper

LangGraph integration

Slack approval bot

Variant 2: Post-detection review

Configure the custom scorer

Webhook receiver: post-detection review ticket

High-risk tool classification

Custom scorer for dangerous tool detection

Compliance value

OWASP LLM06 — Excessive Agency

EU AI Act Article 14 — Human Oversight

SOC 2 CC6.1 — Logical Access Controls

Operational considerations

Latency

Approval fatigue

Failure mode: approval service unavailable