Implementation:Openai Openai agents python Runner Resume Pattern

Overview

This page documents the concrete pattern for resuming an interrupted agent run by passing a RunState back to Runner.run(). This is a Pattern Doc showing the complete end-to-end workflow for human-in-the-loop tool approval.

Runner.run() Signature

Defined in src/agents/run.py (lines 154-231):

class Runner:
    @classmethod
    async def run(
        cls,
        starting_agent: Agent[TContext],
        input: str | list[TResponseInputItem] | RunState[TContext],
        *,
        context: TContext | None = None,
        max_turns: int = DEFAULT_MAX_TURNS,
        hooks: RunHooks[TContext] | None = None,
        run_config: RunConfig | None = None,
        error_handlers: RunErrorHandlers[TContext] | None = None,
        previous_response_id: str | None = None,
        auto_previous_response_id: bool = False,
        conversation_id: str | None = None,
        session: Session | None = None,
    ) -> RunResult:

The input parameter accepts RunState[TContext] as one of its union types. When a RunState is passed, the runner restores the execution context from the state and resumes from the interruption point.

Import

from agents import Agent, Runner

Core Pattern: The Approval Loop

# 1. Initial run
result = await Runner.run(agent, "Do something dangerous")

# 2. Check for interruptions and loop until none remain
while result.interruptions:
    state = result.to_state()

    # 3. Process each pending approval
    for item in state.get_interruptions():
        state.approve(item)  # or state.reject(item)

    # 4. Resume execution with the updated state
    result = await Runner.run(agent, state)

# 5. All approvals handled; use the final result
print(result.final_output)

Step-by-step Explanation

1. Initial run: Start the agent with a user message. If no tools require approval, the loop is skipped entirely.
2. Check for interruptions: The while loop checks result.interruptions. This list is non-empty only when tool calls are pending approval.
3. Process approvals: Iterate over pending ToolApprovalItem objects and approve or reject each one.
4. Resume execution: Pass the RunState (with recorded decisions) back to Runner.run(). The run continues from where it stopped. If the resumed run produces further interruptions, the loop repeats.
5. Final result: When no more interruptions remain, the final output is available.

Full Example: Shell Tool with Approval

from agents import Agent, Runner, ShellTool

async def my_shell_executor(request):
    import asyncio
    proc = await asyncio.create_subprocess_shell(
        " ".join(request.data.action.commands),
        stdout=asyncio.subprocess.PIPE,
        stderr=asyncio.subprocess.PIPE,
    )
    stdout, stderr = await proc.communicate()
    return stdout.decode()

shell = ShellTool(executor=my_shell_executor, needs_approval=True)
agent = Agent(
    name="admin",
    instructions="Run shell commands when asked.",
    tools=[shell],
)

# Run with iterative approval
result = await Runner.run(agent, "List files in /tmp")

while result.interruptions:
    state = result.to_state()
    for item in state.get_interruptions():
        print(f"Tool '{item.tool_name}' wants to execute. Approve? ", end="")
        decision = input("(y/n): ")
        if decision.lower() == "y":
            state.approve(item)
        else:
            state.reject(item)
    result = await Runner.run(agent, state)

print(result.final_output)

Example: Mixed Approval and Rejection

from agents import Agent, Runner, function_tool

@function_tool(needs_approval=True)
def read_file(path: str) -> str:
    """Read a file."""
    with open(path) as f:
        return f.read()

@function_tool(needs_approval=True)
def delete_file(path: str) -> str:
    """Delete a file."""
    import os
    os.remove(path)
    return f"Deleted {path}"

agent = Agent(
    name="file_agent",
    instructions="Manage files as requested.",
    tools=[read_file, delete_file],
)

result = await Runner.run(agent, "Read config.txt and delete temp.log")

while result.interruptions:
    state = result.to_state()
    for item in state.get_interruptions():
        if item.tool_name == "read_file":
            state.approve(item)
        elif item.tool_name == "delete_file":
            state.reject(item)
    result = await Runner.run(agent, state)

print(result.final_output)

In this example, read_file calls are approved while delete_file calls are rejected. The model receives an error message for the rejected call and can adapt its response accordingly.

Example: Using always_approve for Trusted Tools

result = await Runner.run(agent, "Process the data pipeline")

while result.interruptions:
    state = result.to_state()
    for item in state.get_interruptions():
        if item.tool_name == "read_file":
            # Trust all future read_file calls in this run
            state.approve(item, always_approve=True)
        else:
            state.approve(item)
    result = await Runner.run(agent, state)

After the first read_file approval with always_approve=True, subsequent read_file calls will not produce interruptions and will execute automatically.

Key Behavioral Notes

• Turn counter is not incremented by resumption. Only actual model calls advance the turn counter.
• Rejected tools receive an error message in the model's input, allowing the model to adapt.
• Multiple resume cycles are supported. Each cycle may produce new interruptions.
• The same RunState accumulates approval decisions across cycles via the RunContextWrapper._approvals dictionary.

Source References

• src/agents/run.py lines 154-231: Runner.run() accepting RunState