Implement Human-in-the-Loop Approval Workflows for Tool Execution in openai-agents-python

To implement human-in-the-loop approval workflows in openai-agents-python, mark tools with needs_approval=True, handle the resulting ToolApprovalItem interruptions via RunState.approve() or RunState.reject(), and resume the run to complete execution.

The openai-agents-python library provides native support for human-in-the-loop (HITL) approval workflows, allowing developers to pause agent execution when sensitive tools are invoked. This capability is essential for production systems where autonomous actions require explicit human authorization before execution.

Marking Tools for Approval

The foundation of any human-in-the-loop workflow begins at the tool definition level. In src/agents/tool.py, the function_tool decorator accepts a needs_approval parameter that signals the runtime to gate execution.

Static Approval Requirements

Set needs_approval=True to require approval for every invocation:

from agents import function_tool, Agent, Runner

@function_tool(name_override="delete_database", needs_approval=True)
def delete_database(database_name: str) -> str:
    """Permanently delete a database. Requires human approval."""
    return f"Database {database_name} deleted"

The decorator stores this metadata in the tool definition. When the model generates a function call for delete_database, the runtime checks this flag before executing the underlying Python function.

Dynamic Approval Requirements

For conditional approval logic, pass a callable or awaitable to needs_approval:

def requires_approval_for_production(context, tool_call) -> bool:
    """Only require approval in production environments."""
    return context.get("environment") == "production"

@function_tool(
    name_override="modify_user",
    needs_approval=requires_approval_for_production
)
def modify_user(user_id: str, changes: dict) -> str:
    """Modify user account settings."""
    return f"Updated user {user_id}"

The runtime evaluates this function via evaluate_needs_approval_setting in src/agents/util/_approvals.py, passing the current run context and tool call details to determine if the interruption should occur.

Handling ToolApprovalItem Interruptions

When a tool requiring approval is invoked, the runtime pauses execution and surfaces a ToolApprovalItem. In src/agents/run_internal/turn_resolution.py, the turn-resolution logic detects pending approvals at line 889, creating the interruption object instead of executing the tool.

The ToolApprovalItem class, defined in src/agents/items.py at line 480, encapsulates:

• tool_name: The identifier of the invoked tool
• arguments: The parameters passed by the model
• tool_call_id: The unique identifier for this specific invocation
• namespace: The tool's namespace for lookup

When the runtime encounters this item, it returns a RunResult with the interruptions list populated, allowing external code to inspect and resolve the pending operations.

Approving or Rejecting Tool Calls

The RunState class in src/agents/run_state.py provides the public API for resolving interruptions. After receiving a result with pending approvals, you interact with these methods to continue execution.

Using RunState.approve()

To approve a tool call and optionally make the approval sticky for subsequent invocations:

from agents.run_state import RunState

# First run returns interruption

result = await Runner.run(agent, user_input="delete the staging db")
if result.interruptions:
    approval_item = result.interruptions[0]
    
    # Create state from result

    state: RunState = result.to_state()
    
    # Approve this call and remember for this run

    state.approve(approval_item, always_approve=True)
    
    # Resume execution

    final_result = await Runner.run(agent, state)

The approve method updates the internal _approvals map in RunContextWrapper (line 346 in src/agents/run_context.py), recording the decision and clearing the interruption flag.

Using RunState.reject()

To reject a tool call and optionally provide feedback to the model:

state.reject(
    approval_item, 
    always_reject=True, 
    rejection_message="User denied permission to delete databases"
)

resumed = await Runner.run(agent, state)

# The model receives the rejection_message as the tool output

When rejected, the runtime creates a rejection item that feeds back into the conversation history, allowing the model to respond appropriately to the denial.

Realtime Session Implementation

For streaming Realtime sessions, the approval workflow uses the session API defined in src/agents/realtime/session.py. At line 572, the approve_tool_call method provides the streaming equivalent:

session = await RealtimeSession.create(agent, model=model)

# Process until approval needed

await session.run_until_interruption()

# Approve by call_id

await session.approve_tool_call(call_id="call_123", always=True)

# Continue streaming

async for event in session:
    print(event)

This method internally invokes RunContextWrapper.approve_tool and manages the streaming state transition, ensuring that approved calls execute immediately within the Realtime context.

Complete Implementation Example

The test utilities in tests/utils/hitl.py demonstrate the minimal consumer pattern. Here is a production-ready implementation combining all components:

import asyncio
from agents import Agent, Runner, function_tool
from agents.run_state import RunState
from agents.items import ToolApprovalItem

@function_tool(needs_approval=True)
def transfer_funds(amount: float, destination: str) -> str:
    """Transfer funds to external account."""
    return f"Transferred ${amount} to {destination}"

async def main():
    agent = Agent(
        name="finance_agent",
        tools=[transfer_funds],
        instructions="Help users with financial transactions."
    )
    
    # Initial run

    result = await Runner.run(agent, "Send $1000 to account 12345")
    
    # Handle approval workflow

    while result.interruptions:
        approval: ToolApprovalItem = result.interruptions[0]
        print(f"Approval required for: {approval.tool_name}")
        print(f"Arguments: {approval.arguments}")
        
        # In production, this would be a UI prompt

        user_input = input("Approve? (y/n): ")
        
        state: RunState = result.to_state()
        
        if user_input.lower() == 'y':
            state.approve(approval, always_approve=False)
        else:
            state.reject(approval, rejection_message="User denied transfer")
        
        # Resume with updated state

        result = await Runner.run(agent, state)
    
    print(f"Final output: {result.output}")

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

This example demonstrates the complete loop: tool definition with approval requirements, interruption handling, user decision recording, and run resumption.

Summary

Implementing human-in-the-loop approval workflows in openai-agents-python requires understanding these key components:

• Mark tools for approval using needs_approval=True (static) or callable predicates (dynamic) in the function_tool decorator from src/agents/tool.py
• Handle interruptions by checking result.interruptions for ToolApprovalItem objects created by the turn-resolution logic in src/agents/run_internal/turn_resolution.py
• Resolve approvals via RunState.approve() or RunState.reject() from src/agents/run_state.py, which update the internal approval map in src/agents/run_context.py
• Resume execution by passing the updated RunState to Runner.run() for standard agents or using session.approve_tool_call() for Realtime sessions in src/agents/realtime/session.py

Frequently Asked Questions

How do I make tool approvals sticky across multiple calls in the same run?

Use the always_approve=True parameter when calling state.approve(). According to the implementation in src/agents/run_context.py, this stores the approval decision in the per-run _approvals map, causing subsequent invocations of the same tool to bypass the interruption check during turn resolution.

Can I conditionally require approval based on the tool arguments or user context?

Yes. Instead of a boolean, pass a callable or async function to needs_approval in the function_tool decorator. The runtime evaluates this via evaluate_needs_approval_setting in src/agents/util/_approvals.py, passing the current RunContextWrapper and tool call details, allowing you to implement logic such as "only require approval for transfers over $1000" or "skip approval for admin users."

What happens if I reject a tool call instead of approving it?

When you call state.reject() with an optional rejection_message, the runtime creates a rejection output item that gets fed back to the model as the tool's result. As implemented in src/agents/run_state.py, this allows the model to receive explicit feedback about why the action was denied, enabling it to suggest alternatives or ask clarifying questions rather than failing silently.

How do I implement approval workflows in streaming Realtime sessions?

For Realtime sessions, use the approve_tool_call method in src/agents/realtime/session.py at line 572. After calling session.run_until_interruption() to pause on a ToolApprovalItem, call session.approve_tool_call(call_id, always=True) using the tool_call_id from the interruption. This internally invokes the same approval logic as the standard RunState API but manages the streaming session state transition automatically.

{
  "mcpServers": {
    "instagit": {
      "command": "npx",
      "args": ["-y", "instagit@latest"]
    }
  }
}