Implementing Human-in-the-Loop Approval Gates in ADK

Autonomous agents interacting with payment gateways, production databases, or customer communication tools introduce financial, regulatory, and operational risks. An enterprise agent harness must enforce mandatory human sign-off before irreversible actions are dispatched.

Google ADK provides native support for Human-in-the-Loop (HITL) Interrupt Gates. These gates pause execution right before sensitive tool invocation, persist pending context, wait for explicit human authorization, and resume cleanly.

The HITL Lifecycle Architecture

The HITL governance pattern follows four distinct stages:

1. Tool Interception: A before_tool_callback intercepts sensitive tool invocation requests and checks governance rules.
2. State Serialization: If thresholds are breached, tool execution is suspended and session state is saved as AWAITING_HUMAN_SIGN_OFF.
3. Human Review: An external admin portal, Slackbot, or email workflow alerts a human manager with action details.
4. Resume or Cancel: The session is resumed using an approval token or canceled safely without modifying upstream resources.

[!IMPORTANT] Returning False inside an ADK before_tool_callback immediately halts tool execution and leaves session state available for inspection and mutation.

Recipe: Escalation Gate for High-Value Wire Transfers

Below is a complete implementation of a financial agent that intercepts wire transfer requests exceeding $1,000 and pauses execution until authorized:

import asyncio
from google.adk.agents import Agent
from google.adk.tools import FunctionTool
from google.adk.agents.callback_context import CallbackContext
from google.adk.runners import Runner
from google.adk.sessions import InMemorySessionService

# 1. Sensitive Tool Definition
def execute_wire_transfer(recipient_account: str, amount: float) -> dict:
    """Executes a financial wire transfer to a target bank account.

    Args:
        recipient_account: The target bank account number.
        amount: The transfer amount in USD.
    """
    return {
        "status": "COMPLETED",
        "recipient": recipient_account,
        "transferred_amount": amount,
        "transaction_id": "TXN-998124"
    }

wire_tool = FunctionTool(func=execute_wire_transfer)

# 2. HITL Governance Callback
async def hitl_approval_guardrail(tool_name: str, args: dict, callback_context: CallbackContext) -> bool:
    """Before-tool callback that intercepts wire transfers over $1,000."""
    state = callback_context.state

    if tool_name == "execute_wire_transfer":
        amount = args.get("amount", 0.0)
        print(f"[GOVERNANCE CHECK] Wire Transfer Request: ${amount} to {args.get('recipient_account')}")

        # Check for explicit approval token in session state
        if amount > 1000.0 and not state.get("human_approved", False):
            print("[INTERRUPT GATE] Amount exceeds $1,000 threshold! Execution PAUSED for human approval.")
            state["pending_action"] = {"tool": tool_name, "args": args}
            state["status"] = "AWAITING_HUMAN_SIGN_OFF"

            # Returning False cancels/pauses tool execution in ADK
            return False

    return True

# 3. Agent Construction
finance_agent = Agent(
    name="finance_agent",
    model="gemini-2.5-flash",
    instruction="You assist with financial transactions using `execute_wire_transfer`.",
    tools=[wire_tool],
    before_tool_callback=hitl_approval_guardrail
)

# 4. Interactive Simulation
async def main():
    session_service = InMemorySessionService()
    runner = Runner(agent=finance_agent, session_service=session_service)
    session_id = "hitl-demo-session"

    print("--- Test 1: Submitting High-Value Transfer ($5,000) ---")
    res1 = await runner.run_async(
        session_id=session_id,
        message="Please transfer $5000 to account ACCT-7712."
    )

    # Check session state after execution attempt
    session = await session_service.get_session(session_id)
    print("Session Status:", session.state.get("status", "NORMAL"))

    print("\n--- Test 2: Manager Simulates Approval Sign-off ---")
    # Simulate external approval signal mutating session state
    session.state["human_approved"] = True
    session.state["status"] = "APPROVED"
    await session_service.save_session(session)

    res2 = await runner.run_async(
        session_id=session_id,
        message="Manager has approved the transfer. Please proceed."
    )
    print("Agent Response:\n", res2.text)

if __name__ == "__main__":
    asyncio.run(main())

How the Guardrail Operates

1. Triggering the Callback: The before_tool_callback fires immediately when Gemini generates a tool call to execute_wire_transfer.
2. Threshold Verification: The callback inspects args["amount"]. If the requested transfer is $> $1,000$ and no human_approved flag exists in callback_context.state, it returns False.
3. Execution Pause: Returning False prevents the Python function from running. The pending tool arguments are saved directly into the session state dictionary.
4. State Mutation & Approval: An external manager interface modifies session.state["human_approved"] = True and saves the updated session.
5. Clean Resume: On the subsequent turn, the callback re-evaluates the condition, confirms approval, and allows the tool invocation to proceed safely.

Production Implementation Checklist

[!WARNING] In production systems, never rely on in-memory storage for pending human approvals.

• Persistent Session Storage: Use DatabaseSessionService (PostgreSQL/Spanner) so paused states survive server restarts and blue-green deployments.
• Audit Trails: Write pending actions to an immutable compliance ledger before returning False.
• Approval SLA & Timeouts: Set automatic expiration limits on pending states (e.g., auto-cancel if unapproved after 24 hours).
• Notification Webhooks: Trigger PagerDuty, Slack, or Teams alerts with direct approval links when an interrupt gate is hit.

Key Takeaway

The before_tool_callback is your ultimate governance control point in ADK. Use it to build secure HITL workflows that protect business assets while granting AI agents full autonomy within safe limits.