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

> Learn to implement human-in-the-loop approval for tool execution in openai-agents-python. Approve or reject tool calls, control execution flow, and ensure safe AI interactions.

- Repository: [OpenAI/openai-agents-python](https://github.com/openai/openai-agents-python)
- Tags: how-to-guide
- Published: 2026-04-17

---

**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`](https://github.com/openai/openai-agents-python/blob/main/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:

```python
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`:

```python
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`](https://github.com/openai/openai-agents-python/blob/main/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`](https://github.com/openai/openai-agents-python/blob/main/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`](https://github.com/openai/openai-agents-python/blob/main/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`](https://github.com/openai/openai-agents-python/blob/main/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:

```python
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`](https://github.com/openai/openai-agents-python/blob/main/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:

```python
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`](https://github.com/openai/openai-agents-python/blob/main/src/agents/realtime/session.py). At line 572, the `approve_tool_call` method provides the streaming equivalent:

```python
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`](https://github.com/openai/openai-agents-python/blob/main/tests/utils/hitl.py) demonstrate the minimal consumer pattern. Here is a production-ready implementation combining all components:

```python
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`](https://github.com/openai/openai-agents-python/blob/main/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`](https://github.com/openai/openai-agents-python/blob/main/src/agents/run_internal/turn_resolution.py)
- **Resolve approvals** via `RunState.approve()` or `RunState.reject()` from [`src/agents/run_state.py`](https://github.com/openai/openai-agents-python/blob/main/src/agents/run_state.py), which update the internal approval map in [`src/agents/run_context.py`](https://github.com/openai/openai-agents-python/blob/main/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`](https://github.com/openai/openai-agents-python/blob/main/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`](https://github.com/openai/openai-agents-python/blob/main/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`](https://github.com/openai/openai-agents-python/blob/main/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`](https://github.com/openai/openai-agents-python/blob/main/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`](https://github.com/openai/openai-agents-python/blob/main/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.