Skip to main content

​
Overview

Human-in-the-loop (HITL) means requiring human approval before AI agents perform certain actions. This is critical for:
  • High-stakes operations: Deleting data, making purchases, sending communications
  • Compliance requirements: Regulated industries requiring human oversight
  • Safety guardrails: Preventing unintended consequences in production
  • Gradual trust building: Starting with oversight, removing it as confidence grows
The key insight with Agentica is that you don’t need special framework features for HITL. It’s just wrapping the functions you pass to agents with approval logic. Agentica makes this trivial because agents just use regular functions—wrap those functions, and you have human oversight.
What’s in this guide:

​
The Core Pattern

Instead of passing a function directly to an agent, wrap it with approval logic: 
from agentica import spawn

# Original function
def delete_file(path: str) -> None:
    os.remove(path)

# Wrapped version with approval
def delete_file_with_approval(path: str) -> None:
    # Ask for approval however you want
    response = input(f"Delete {path}? (yes/no): ")
    if response.lower() != 'yes':
        raise PermissionError(f"User denied deletion of {path}")

    # If approved, call the original function
    os.remove(path)

# Pass the wrapped version to the agent
agent = await spawn(premise="You are a file organizer")
await agent.call(None, "Clean up the temp directory", delete_file=delete_file_with_approval)
No middleware, no special API, no framework hooks. Just wrap the function you’re passing.
The agent sees the approval prompt as part of the function’s behavior. If you deny the action, the agent receives an exception and can adapt accordingly.

​
Decision Types

When reviewing an action, you have three typical options:
DecisionDescriptionWhen to UseAgent Response
ApproveExecute the action exactly as proposedAction is correct and safeProceeds with execution
EditModify arguments before executingAction is almost right but needs tweakingExecutes with your changes
RejectDeny the action with feedbackAction is wrong or unsafeSees your feedback and adapts strategy
response = input(f"Approve action? (yes/no): ")
if response.lower() == 'yes':
    return original_function(*args, **kwargs)
else:
    raise PermissionError("Action denied")
Modify the agent’s proposed arguments before executing. This is useful when the action is almost right but needs adjustment:
def send_email_with_editable_approval(to: str, subject: str, body: str) -> str:
    print()
    print(f"📧 Agent wants to send email:")
    print(f"   To: {to}")
    print(f"   Subject: {subject}")
    print(f"   Body: {body[:100]}...")

    print()
    response = input("   Action? (approve/edit/reject): ")

    if response == 'approve':
        return send_email(to, subject, body)
    elif response == 'edit':
        # Let human modify the arguments
        new_to = input(f"   To [{to}]: ") or to
        new_subject = input(f"   Subject [{subject}]: ") or subject
        new_body = input(f"   Body (press enter to keep current): ") or body

        print(f"Executing with modified arguments")
        return send_email(new_to, new_subject, new_body)
    else:
        raise PermissionError("Email sending denied")
Deny the action but provide feedback that helps the agent understand why and try a different approach:
def execute_sql_with_feedback(query: str) -> list:
    print(f"\nAgent wants to execute SQL:")
    print(f"   Query: {query[:100]}...")

    response = input("\n   Approve? (yes/no): ")

    if response.lower() == 'yes':
        return run_sql_query(query)
    else:
        # Provide specific feedback to help the agent
        feedback = input("   Reason for rejection: ")
        raise PermissionError(
            f"SQL execution denied. Feedback: {feedback}. "
            f"Please revise the query and try again."
        )
        # The agent sees this error message and can adapt its approach
When you raise an exception with a descriptive message, the agent sees that message and can adjust its strategy. For example, if you reject a query because “must use READ ONLY transaction,” the agent might then add that to its next attempt.

​
Approval Mechanisms

How you ask for approval depends on your application context. Here are practical patterns:
Perfect for development, scripts, and command-line tools:
def with_cli_approval(func, action_description: str):
    """Decorator that adds CLI approval to any function."""
    def wrapper(*args, **kwargs):
        print(f"\n🤖 Agent wants to: {action_description}")
        print(f"   Args: {args}, Kwargs: {kwargs}")
        response = input("   Approve? (yes/no): ")

        if response.lower() != 'yes':
            raise PermissionError(f"User denied: {action_description}")

        return func(*args, **kwargs)
    return wrapper

# Use it
send_email = with_cli_approval(
    original_send_email,
    "send an email"
)

agent = await spawn(premise="You are an assistant")
await agent.call(None, "Email the team about the meeting", send_email=send_email)
For web applications running in the browser:
function withBrowserApproval<T extends (...args: any[]) => any>(
  func: T,
  actionDescription: string
): T {
  return ((...args: any[]) => {
    const confirmed = window.confirm(
      `Agent wants to ${actionDescription}\n\n` +
      `Details: ${JSON.stringify(args, null, 2)}\n\n` +
      `Allow this action?`
    );

    if (!confirmed) {
      throw new Error(`User denied: ${actionDescription}`);
    }

    return func(...args);
  }) as T;
}

// Use it with Slack integration
const postMessage = withBrowserApproval(
  slackClient.chat.postMessage.bind(slackClient.chat),
  "post a Slack message"
);

const agent = await spawn({ premise: "You are a Slack assistant" });
await agent.call(
  "Send a message to #engineering about the deploy",
  { postMessage }
);
For production systems with approval workflows, you may want approvals to go through a proper web interface where managers can review requests, see context, and make informed decisions. This is especially useful when:
  • The person approving isn’t at their terminal
  • You need a record of who approved what and when
  • Multiple people might need to approve the same action
  • You want to batch approval requests for review
The pattern: create an approval request via HTTP, poll until approved/denied, then proceed or raise an error.
import requests
import time

def with_dashboard_approval(func, action_description: str, dashboard_url: str):
    """Request approval via web dashboard."""
    def wrapper(*args, **kwargs):
        # Create approval request
        response = requests.post(
            f"{dashboard_url}/api/approvals",
            json={"action": action_description, "args": list(args)}
        )
        request_id = response.json()["id"]
        print(f"⏳ Waiting for approval at {dashboard_url}/approvals/{request_id}")

        # Poll for approval
        while True:
            status = requests.get(f"{dashboard_url}/api/approvals/{request_id}").json()

            if status["approved"]:
                print(f"✅ Action approved")
                return func(*args, **kwargs)
            elif status["denied"]:
                raise PermissionError(f"Action denied: {status['reason']}")

            time.sleep(2)  # Poll every 2 seconds
    return wrapper
See the Customer Support example below for a complete production implementation using this pattern.
Request approval via chat platforms where your team already is:
from slack_sdk import WebClient

def with_slack_approval(func, action_description: str, slack_client: WebClient, channel: str):
    """Request approval via Slack message with buttons."""
    def wrapper(*args, **kwargs):
        # Send approval request with interactive buttons
        response = slack_client.chat_postMessage(
            channel=channel,
            text=f"🤖 Agent approval needed: {action_description}",
            blocks=[
                {
                    "type": "section",
                    "text": {
                        "type": "mrkdwn",
                        "text": f"*Agent wants to:* {action_description}\n*Details:* ```{args}```"
                    }
                },
                {
                    "type": "actions",
                    "elements": [
                        {
                            "type": "button",
                            "text": {"type": "plain_text", "text": "Approve"},
                            "style": "primary",
                            "action_id": "approve_action"
                        },
                        {
                            "type": "button",
                            "text": {"type": "plain_text", "text": "Deny"},
                            "style": "danger",
                            "action_id": "deny_action"
                        }
                    ]
                }
            ]
        )

        # Wait for button click (would need webhook handler in real implementation)
        # This is simplified - production needs proper Slack event handling
        approval_status = wait_for_slack_response(response['ts'])

        if approval_status != 'approved':
            raise PermissionError(f"Action denied via Slack")

        return func(*args, **kwargs)

    return wrapper
Only require approval for high-risk operations:
def with_conditional_approval(func, should_require_approval):
    """Only require approval when condition is met."""
    def wrapper(*args, **kwargs):
        # Check if this specific call needs approval
        if should_require_approval(*args, **kwargs):
            response = input(f"⚠️  High-risk operation detected. Approve? (yes/no): ")
            if response.lower() != 'yes':
                raise PermissionError("User denied high-risk operation")

        return func(*args, **kwargs)
    return wrapper

# Only approve transfers over $10,000
def needs_approval_for_large_transfer(amount: float, recipient: str) -> bool:
    return amount > 10000

transfer_money = with_conditional_approval(
    bank.transfer,
    needs_approval_for_large_transfer
)

agent = await spawn(premise="Financial assistant")
await agent.call(None, "Pay the invoices", transfer_money=transfer_money)

​
Practical Examples

An agent that can deploy code, but requires approval for production:
from agentica import spawn
from dataclasses import dataclass

@dataclass
class DeploymentConfig:
    environment: str
    branch: str
    services: list[str]

def deploy_to_environment(config: DeploymentConfig) -> str:
    """Actually perform the deployment."""
    # Deployment logic here
    return f"Deployed {config.services} to {config.environment}"

def deploy_with_approval(config: DeploymentConfig) -> str:
    """Wrapper that requires approval for production."""
    if config.environment == "production":
        print(f"\n⚠️  PRODUCTION DEPLOYMENT REQUEST")
        print(f"   Branch: {config.branch}")
        print(f"   Services: {', '.join(config.services)}")

        response = input("\n   Deploy to production? (yes/no): ")
        if response.lower() != 'yes':
            raise PermissionError("Production deployment denied")

        print("\n✅ Production deployment approved")

    return deploy_to_environment(config)

# Use with agent
agent = await spawn(
    premise="You are a DevOps assistant. Deploy services when requested.",
    model="openai:gpt-4.1"
)

result = await agent.call(
    str,
    "Deploy the auth-service from main branch to production",
    deploy=deploy_with_approval
)
# Agent will call deploy_with_approval, which prompts for confirmation
Agent handles support tickets but requires approval for refunds:
from agentica import spawn

class SupportTools:
    def __init__(self, approval_dashboard_url: str):
        self.dashboard_url = approval_dashboard_url

    def send_canned_response(self, ticket_id: str, response_type: str) -> str:
        """Automated responses don't need approval."""
        return f"Sent {response_type} response to ticket {ticket_id}"

    def issue_refund(self, ticket_id: str, amount: float, reason: str) -> str:
        """Refunds require approval via dashboard."""
        # Create approval request
        response = requests.post(
            f"{self.dashboard_url}/api/refund-approvals",
            json={
                "ticket_id": ticket_id,
                "amount": amount,
                "reason": reason
            }
        )
        approval_id = response.json()["id"]

        print(f"💰 Refund approval needed: ${amount} for ticket {ticket_id}")
        print(f"   Review at: {self.dashboard_url}/approvals/{approval_id}")

        # Wait for approval
        while True:
            status = requests.get(
                f"{self.dashboard_url}/api/refund-approvals/{approval_id}"
            ).json()

            if status["approved"]:
                # Actually process the refund
                return f"Refunded ${amount} to customer (ticket {ticket_id})"
            elif status["denied"]:
                raise PermissionError(f"Refund denied: {status['denial_reason']}")

            time.sleep(3)

# Use with support agent
tools = SupportTools("https://support-dashboard.company.com")

agent = await spawn(
    premise="""You are a customer support agent.
    You can send canned responses for common issues.
    For refund requests, use issue_refund - it will be reviewed by a manager."""
)

result = await agent.call(
    str,
    "Handle ticket #12345 - customer unhappy with product quality, requesting refund of $49.99",
    send_response=tools.send_canned_response,
    issue_refund=tools.issue_refund
)
Agent can query data freely, but needs approval for modifications:
from agentica import spawn
import pandas as pd

class DatabaseTools:
    def __init__(self, db_connection):
        self.db = db_connection

    def query_data(self, sql: str) -> dict:
        """Read-only queries are always allowed."""
        if not sql.strip().upper().startswith('SELECT'):
            raise ValueError("Only SELECT queries allowed in query_data")

        result = pd.read_sql(sql, self.db)
        return result.to_dict()

    def delete_records(self, table: str, condition: str) -> str:
        """Deletions require approval."""
        # Show what would be deleted
        preview_sql = f"SELECT * FROM {table} WHERE {condition}"
        preview = pd.read_sql(preview_sql, self.db)

        print(f"\n⚠️  DELETION REQUEST")
        print(f"   Table: {table}")
        print(f"   Condition: {condition}")
        print(f"   Rows to be deleted: {len(preview)}")
        print(f"\n   Preview:")
        print(preview.head().to_string())

        response = input(f"\n   Delete these {len(preview)} rows? (yes/no): ")
        if response.lower() != 'yes':
            raise PermissionError("Deletion denied")

        # Perform deletion
        self.db.execute(f"DELETE FROM {table} WHERE {condition}")
        return f"Deleted {len(preview)} rows from {table}"

# Use with data analysis agent
db_tools = DatabaseTools(db_connection)

agent = await spawn(
    premise="You are a data analyst. Query data to answer questions."
)

await agent.call(
    None,
    "Find and delete all test users created before 2024",
    query_data=db_tools.query_data,
    delete_records=db_tools.delete_records
)
# Agent can query freely, but deletions prompt for approval

​
Advanced Patterns

When an agent wants to perform multiple actions at once, you can collect them all, review them together, then decide on each one.How it works:
  1. When the agent calls a wrapped function, instead of executing immediately, the call is queued and a placeholder like "PENDING_APPROVAL_1" is returned to the agent
  2. The agent continues executing and may queue multiple actions
  3. After the agent finishes, you call review_and_execute() to see all pending actions at once
  4. You can approve all, deny all, or handle each individually
  5. The actual functions execute based on your decisions
This pattern is useful when you want to see the full scope of what the agent plans to do before committing to any individual action.
def create_batch_approver(*funcs_with_names):
    """
    Create wrapper that shows all pending actions for review.
    funcs_with_names: list of (name, function) tuples
    """
    pending_actions = []

    def make_wrapper(name, func):
        def wrapper(*args, **kwargs):
            # Collect action for later approval
            action_id = len(pending_actions) + 1
            pending_actions.append({
                'name': name,
                'func': func,
                'args': args,
                'kwargs': kwargs,
                'id': action_id
            })
            # Return marker that agent can see
            return f"PENDING_APPROVAL_{action_id}"
        return wrapper

    # Create wrapped versions
    wrapped_funcs = {
        name: make_wrapper(name, func)
        for name, func in funcs_with_names
    }
    
    # Store results for wait_for_approvals to return
    approval_results = {}
    approval_complete = {'done': False}

    def wait_for_approvals():
        """
        Agent calls this to block until human reviews all pending actions.
        Returns dict mapping action IDs to their results.
        """
        import time
        print("\n[Agent is waiting for approval of pending actions...]")
        
        # Block until review_and_execute is called
        while not approval_complete['done']:
            time.sleep(0.1)
        
        # Return results to agent
        approval_complete['done'] = False  # Reset for next batch
        return approval_results.copy()

    def review_and_execute():
        """Review all pending actions and execute based on decisions."""
        if not pending_actions:
            return []

        print(f"\nAgent wants to perform {len(pending_actions)} actions:")

        # Show all actions
        for i, action in enumerate(pending_actions, 1):
            print(f"\n{i}. {action['name']}")
            print(f"   Args: {action['args']}")
            print(f"   Kwargs: {action['kwargs']}")

        # Get batch decision
        print(f"\nOptions: 'all' (approve all), 'none' (deny all), 'review' (individual), 'edit N' (edit action N)")
        batch_decision = input("Decision: ").strip().lower()

        results = []
        
        if batch_decision == 'all':
            # Approve all
            for action in pending_actions:
                result = action['func'](*action['args'], **action['kwargs'])
                results.append(result)
                approval_results[action['id']] = result
                print(f"✓ Executed {action['name']}: {result}")
                
        elif batch_decision == 'none':
            # Deny all
            feedback = input("Reason for denying all: ")
            for action in pending_actions:
                rejection = f"REJECTED: {feedback}"
                results.append(rejection)
                approval_results[action['id']] = rejection
            print(f"✗ Denied all {len(pending_actions)} actions")
            
        elif batch_decision.startswith('edit '):
            # Edit specific action
            try:
                idx = int(batch_decision.split()[1]) - 1
                if 0 <= idx < len(pending_actions):
                    action = pending_actions[idx]
                    print(f"\nEditing action {idx + 1}: {action['name']}")
                    print(f"   Current args: {action['args']}")
                    
                    # Simple example: modify first positional arg
                    if action['args']:
                        new_arg = input(f"   {action['args'][0]} -> ")
                        if new_arg:
                            modified_args = (new_arg,) + action['args'][1:]
                            action['args'] = modified_args
                            print(f"   Updated. Run review_and_execute() again to approve.")
                    return []  # Return empty, user needs to review again
                else:
                    print(f"Invalid index. Use 1-{len(pending_actions)}")
                    return []
            except (ValueError, IndexError):
                print("Invalid edit command. Use 'edit N' where N is the action number.")
                return []
                
        else:  # 'review' or default - individual review
            for i, action in enumerate(pending_actions, 1):
                print(f"\n--- Action {i}/{len(pending_actions)}: {action['name']} ---")
                print(f"    Args: {action['args']}")
                decision = input("    Decision (approve/reject/skip): ").strip().lower()

                if decision == 'approve':
                    result = action['func'](*action['args'], **action['kwargs'])
                    results.append(result)
                    approval_results[action['id']] = result
                    print(f"    ✓ Executed: {result}")
                elif decision == 'reject':
                    feedback = input("    Reason: ")
                    rejection = f"REJECTED: {feedback}"
                    results.append(rejection)
                    approval_results[action['id']] = rejection
                    print(f"    ✗ Rejected")
                else:  # skip
                    skipped = "SKIPPED"
                    results.append(skipped)
                    approval_results[action['id']] = skipped
                    print(f"    - Skipped")

        pending_actions.clear()
        approval_complete['done'] = True  # Signal to wait_for_approvals
        return results

    wrapped_funcs['_review_and_execute'] = review_and_execute
    wrapped_funcs['wait_for_approvals'] = wait_for_approvals
    return wrapped_funcs

# Usage example - Agent waits for approval
from agentica import spawn

def delete_user(user_id: str) -> str:
    return f"Deleted user {user_id}"

def send_notification(user_id: str, message: str) -> str:
    return f"Sent '{message}' to {user_id}"

# Create batch-approved versions
tools = create_batch_approver(
    ('delete_user', delete_user),
    ('send_notification', send_notification)
)

agent = await spawn(premise="You are a user admin")

# Agent queues actions then waits for approval
await agent.call(
    None,
    """Delete user 123 and notify user 456 about it.
    After queuing these actions, call wait_for_approvals() to get the actual results.""",
    delete_user=tools['delete_user'],
    send_notification=tools['send_notification'],
    wait_for_approvals=tools['wait_for_approvals']
)
# What happens:
# 1. Agent calls delete_user(123) → gets "PENDING_APPROVAL_1"
# 2. Agent calls send_notification(456, "...") → gets "PENDING_APPROVAL_2"  
# 3. Agent calls wait_for_approvals() → BLOCKS waiting for human
# 4. Human reviews in another terminal/thread with tools['_review_and_execute']()
# 5. wait_for_approvals() returns {1: "Deleted user 123", 2: "Sent '...' to 456"}
# 6. Agent can now see which actions succeeded vs were rejected
Key insight: By providing wait_for_approvals() to the agent, the agent can:
  1. Queue up actions (getting placeholders back like "PENDING_APPROVAL_1")
  2. Call wait_for_approvals() to block until human review
  3. Receive a dict mapping action IDs to actual results: {1: "Deleted user 123", 2: "REJECTED: ..."}
  4. Adapt based on what was approved vs rejected
The agent can check results like: if "REJECTED" in results[1] to handle denied actions differently.Alternative: If you don’t give the agent wait_for_approvals(), it will receive placeholders and complete without knowing the real outcomes. Useful when the agent doesn’t need to adapt based on approval decisions.You can customize this pattern however you want: web UI for review, role-based approvals, risk-level grouping, timeouts—just code it.
Example: Agent adapts based on approval results
# Give agent ability to check what was approved
agent = await spawn(premise="You are a deployment manager")

await agent.call(
    str,
    """Deploy services A, B, and C to production.
    After queuing deployments, call wait_for_approvals() to see which were approved.
    For any rejected deployments, log the reason and continue with approved ones.""",
    deploy_service=tools['deploy_service'],
    log_message=tools['log_message'],
    wait_for_approvals=tools['wait_for_approvals']
)

# Agent might do:
# 1. deploy_service("A") → "PENDING_APPROVAL_1"
# 2. deploy_service("B") → "PENDING_APPROVAL_2"
# 3. deploy_service("C") → "PENDING_APPROVAL_3"
# 4. results = wait_for_approvals() → {1: "Deployed A", 2: "REJECTED: staging failed", 3: "Deployed C"}
# 5. log_message("Services A and C deployed. Service B rejected: staging failed")
# 
# Agent adapts: it sees B was rejected and logs accordingly!
Allow a certain number of operations without approval, then require oversight:
class ApprovalBudget:
    def __init__(self, free_operations: int):
        self.remaining = free_operations

    def with_budget(self, func, action_description: str):
        def wrapper(*args, **kwargs):
            if self.remaining > 0:
                self.remaining -= 1
                print(f"✓ Auto-approved ({self.remaining} remaining in budget)")
                return func(*args, **kwargs)

            # Budget exhausted, require approval
            response = input(f"Budget exhausted. Approve '{action_description}'? (yes/no): ")
            if response.lower() != 'yes':
                raise PermissionError(f"Denied: {action_description}")

            return func(*args, **kwargs)
        return wrapper

# Allow 5 free API calls, then require approval
budget = ApprovalBudget(free_operations=5)

make_api_call = budget.with_budget(
    external_api.call,
    "make external API call"
)
Require approval during business hours, allow automation off-hours:
from datetime import datetime

def with_business_hours_approval(func, action_description: str):
    def wrapper(*args, **kwargs):
        hour = datetime.now().hour
        is_business_hours = 9 <= hour < 17

        if is_business_hours:
            # During work hours, require approval
            response = input(f"[Business hours] Approve '{action_description}'? (yes/no): ")
            if response.lower() != 'yes':
                raise PermissionError(f"Denied: {action_description}")
        else:
            # Off hours, auto-approve but log
            print(f"✓ Auto-approved (off-hours): {action_description}")

        return func(*args, **kwargs)
    return wrapper
Log all approval decisions for compliance:
import json
from datetime import datetime

class AuditedApproval:
    def __init__(self, audit_log_path: str):
        self.audit_log = audit_log_path

    def with_audit(self, func, action_description: str):
        def wrapper(*args, **kwargs):
            response = input(f"Approve '{action_description}'? (yes/no): ")
            approved = response.lower() == 'yes'

            # Log the decision
            audit_entry = {
                "timestamp": datetime.now().isoformat(),
                "action": action_description,
                "args": str(args),
                "approved": approved,
                "approver": os.getenv("USER", "unknown")
            }

            with open(self.audit_log, 'a') as f:
                f.write(json.dumps(audit_entry) + '\n')

            if not approved:
                raise PermissionError(f"Denied: {action_description}")

            result = func(*args, **kwargs)

            # Log the result
            result_entry = {
                **audit_entry,
                "completed": True,
                "result": str(result)[:200]  # Truncate long results
            }
            with open(self.audit_log, 'a') as f:
                f.write(json.dumps(result_entry) + '\n')

            return result
        return wrapper

​
Why This Approach Works

Key insight: Agentica agents call real functions, not schema definitions. This means HITL is just wrapping functions—no special framework features needed.
Traditional AI frameworks require special middleware or plugins for human-in-the-loop because they treat tools as schema definitions, not actual code. You configure tools through JSON or decorators, and the framework manages execution. Agentica is different: agents call real functions. When you pass delete_file to an agent, the agent literally calls your delete_file function through RPC. There’s no schema layer, no tool registry, no execution middleware. This means:
  • You control execution — wrap functions however you want
  • Use any approval mechanism — terminal, web, Slack, custom systems
  • Standard programming patterns — decorators, higher-order functions, classes
  • No framework lock-in — approval logic is in your code, not the framework
You don’t need Agentica to provide HITL because you already have everything you need: functions and control flow.

​
Best Practices

When denying an action, raise an exception with a clear message. The agent sees this and can adapt:
# Good - agent understands why
raise PermissionError("Deployment denied: must deploy to staging first")

# Bad - agent gets generic error
raise Exception("No")
The agent might then try an alternative approach based on the error message.
Not everything needs the same oversight:
  • Low risk: Auto-approve, just log
  • Medium risk: CLI approval for development
  • High risk: Dashboard with multiple approvers
  • Critical: Require manager approval + audit trail
Use environment variables to control approval requirements:
def deploy_with_approval(config: DeploymentConfig) -> str:
    # Production always requires approval
    # Other environments only if REQUIRE_APPROVAL=true
    needs_approval = (
        config.environment == "production" or
        os.getenv("REQUIRE_APPROVAL") == "true"
    )

    if needs_approval:
        # ... approval logic
        pass

    return deploy(config)
Don’t wait forever for approval:
import asyncio

async def with_timeout_approval(func, action_description: str, timeout_seconds: int = 300):
    async def wrapper(*args, **kwargs):
        # Start approval request in background
        approval_task = asyncio.create_task(
            get_approval_async(action_description, args)
        )

        try:
            # Wait for approval with timeout
            approved = await asyncio.wait_for(approval_task, timeout=timeout_seconds)
            if not approved:
                raise PermissionError(f"Denied: {action_description}")
        except asyncio.TimeoutError:
            raise TimeoutError(f"Approval timeout for: {action_description}")

        return func(*args, **kwargs)
    return wrapper

​
Next Steps

Best Practices

Production-ready patterns and security

Multi-Agent Systems

Apply HITL in multi-agent workflows

Error Handling

Handle approval denials and failures

Examples

Explore examples & get inspired