feat: #636 Add human-in-the-loop (HITL) support

.gitignore

@@ -149,3 +149,5 @@ cython_debug/
 
 # Redis database files
 dump.rdb
+
+tmp/

AGENTS.md

@@ -55,6 +55,24 @@ The OpenAI Agents Python repository provides the Python Agents SDK, examples, an
 - `.github/PULL_REQUEST_TEMPLATE/pull_request_template.md`: Pull request template to use when opening PRs.
 - `site/`: Built documentation output.
 
+### Agents Core Runtime Guidelines
+
+- `src/agents/run.py` is the runtime entrypoint (`Runner`, `AgentRunner`). Keep it focused on orchestration and public flow control. Put new runtime logic under `src/agents/run_internal/` and import it into `run.py`.
+- When `run.py` grows, refactor helpers into `run_internal/` modules (for example `run_loop.py`, `turn_resolution.py`, `tool_execution.py`, `session_persistence.py`) and leave only wiring and composition in `run.py`.
+- Keep streaming and non-streaming paths behaviorally aligned. Changes to `run_internal/run_loop.py` (`run_single_turn`, `run_single_turn_streamed`, `get_new_response`, `start_streaming`) should be mirrored, and any new streaming item types must be reflected in `src/agents/stream_events.py`.
+- Input guardrails run only on the first turn and only for the starting agent. Resuming an interruption from `RunState` must not increment the turn counter; only actual model calls advance turns.
+- Server-managed conversation (`conversation_id`, `previous_response_id`, `auto_previous_response_id`) uses `OpenAIServerConversationTracker` in `run_internal/oai_conversation.py`. Only deltas should be sent. If `call_model_input_filter` is used, it must return `ModelInputData` with a list input and the tracker must be updated with the filtered input (`mark_input_as_sent`). Session persistence is disabled when server-managed conversation is active.
+- Adding new tool/output/approval item types requires coordinated updates across:
+  - `src/agents/items.py` (RunItem types and conversions)
+  - `src/agents/run_internal/run_steps.py` (ProcessedResponse and tool run structs)
+  - `src/agents/run_internal/turn_resolution.py` (model output processing, run item extraction)
+  - `src/agents/run_internal/tool_execution.py` and `src/agents/run_internal/tool_planning.py`
+  - `src/agents/run_internal/items.py` (normalization, dedupe, approval filtering)
+  - `src/agents/stream_events.py` (stream event names)
+  - `src/agents/run_state.py` (RunState serialization/deserialization)
+  - `src/agents/run_internal/session_persistence.py` (session save/rewind)
+- If the serialized RunState shape changes, bump `CURRENT_SCHEMA_VERSION` in `src/agents/run_state.py` and update serialization/deserialization accordingly.
+
 ## Operation Guide
 
 ### Prerequisites

examples/agent_patterns/agents_as_tools_conditional.py

@@ -2,8 +2,9 @@
 
 from pydantic import BaseModel
 
-from agents import Agent, AgentBase, RunContextWrapper, Runner, trace
-from examples.auto_mode import input_with_fallback
+from agents import Agent, AgentBase, ModelSettings, RunContextWrapper, Runner, trace
+from agents.tool import function_tool
+from examples.auto_mode import confirm_with_fallback, input_with_fallback
 
 """
 This example demonstrates the agents-as-tools pattern with conditional tool enabling.
@@ -26,10 +27,18 @@ def european_enabled(ctx: RunContextWrapper[AppContext], agent: AgentBase) -> bo
     return ctx.context.language_preference == "european"
 
 
+@function_tool(needs_approval=True)
+async def get_user_name() -> str:
+    print("Getting the user's name...")
+    return "Kaz"
+
+
 # Create specialized agents
 spanish_agent = Agent(
     name="spanish_agent",
-    instructions="You respond in Spanish. Always reply to the user's question in Spanish.",
+    instructions="You respond in Spanish. Always reply to the user's question in Spanish. You must call all the tools to best answer the user's question.",
+    model_settings=ModelSettings(tool_choice="required"),
+    tools=[get_user_name],
 )
 
 french_agent = Agent(
@@ -55,6 +64,7 @@ def european_enabled(ctx: RunContextWrapper[AppContext], agent: AgentBase) -> bo
             tool_name="respond_spanish",
             tool_description="Respond to the user's question in Spanish",
             is_enabled=True,  # Always enabled
+            needs_approval=True,  # HITL
         ),
         french_agent.as_tool(
             tool_name="respond_french",
@@ -109,8 +119,24 @@ async def main():
             input=user_request,
             context=context.context,
         )
-
-        print(f"\nResponse:\n{result.final_output}")
+        while result.interruptions:
+
+            async def confirm(question: str) -> bool:
+                return confirm_with_fallback(f"{question} (y/n): ", default=True)
+
+            state = result.to_state()
+            for interruption in result.interruptions:
+                prompt = f"\nDo you approve this tool call: {interruption.name} with arguments {interruption.arguments}?"
+                confirmed = await confirm(prompt)
+                if confirmed:
+                    state.approve(interruption)
+                    print(f"✓ Approved: {interruption.name}")
+                else:
+                    state.reject(interruption)
+                    print(f"✗ Rejected: {interruption.name}")
+            result = await Runner.run(orchestrator, state)
+
+    print(f"\nResponse:\n{result.final_output}")
 
 
 if __name__ == "__main__":

examples/agent_patterns/human_in_the_loop.py

@@ -0,0 +1,137 @@
+"""Human-in-the-loop example with tool approval.
+
+This example demonstrates how to:
+1. Define tools that require approval before execution
+2. Handle interruptions when tool approval is needed
+3. Serialize/deserialize run state to continue execution later
+4. Approve or reject tool calls based on user input
+"""
+
+import asyncio
+import json
+from pathlib import Path
+
+from agents import Agent, Runner, RunState, function_tool
+from examples.auto_mode import confirm_with_fallback
+
+
+@function_tool
+async def get_weather(city: str) -> str:
+    """Get the weather for a given city.
+
+    Args:
+        city: The city to get weather for.
+
+    Returns:
+        Weather information for the city.
+    """
+    return f"The weather in {city} is sunny"
+
+
+async def _needs_temperature_approval(_ctx, params, _call_id) -> bool:
+    """Check if temperature tool needs approval."""
+    return "Oakland" in params.get("city", "")
+
+
+@function_tool(
+    # Dynamic approval: only require approval for Oakland
+    needs_approval=_needs_temperature_approval
+)
+async def get_temperature(city: str) -> str:
+    """Get the temperature for a given city.
+
+    Args:
+        city: The city to get temperature for.
+
+    Returns:
+        Temperature information for the city.
+    """
+    return f"The temperature in {city} is 20° Celsius"
+
+
+# Main agent with tool that requires approval
+agent = Agent(
+    name="Weather Assistant",
+    instructions=(
+        "You are a helpful weather assistant. "
+        "Answer questions about weather and temperature using the available tools."
+    ),
+    tools=[get_weather, get_temperature],
+)
+
+RESULT_PATH = Path(".cache/agent_patterns/human_in_the_loop/result.json")
+
+
+async def confirm(question: str) -> bool:
+    """Prompt user for yes/no confirmation.
+
+    Args:
+        question: The question to ask.
+
+    Returns:
+        True if user confirms, False otherwise.
+    """
+    return confirm_with_fallback(f"{question} (y/n): ", default=True)
+
+
+async def main():
+    """Run the human-in-the-loop example."""
+    result = await Runner.run(
+        agent,
+        "What is the weather and temperature in Oakland?",
+    )
+
+    has_interruptions = len(result.interruptions) > 0
+
+    while has_interruptions:
+        print("\n" + "=" * 80)
+        print("Run interrupted - tool approval required")
+        print("=" * 80)
+
+        # Storing state to file (demonstrating serialization)
+        state = result.to_state()
+        state_json = state.to_json()
+        RESULT_PATH.parent.mkdir(parents=True, exist_ok=True)
+        with RESULT_PATH.open("w") as f:
+            json.dump(state_json, f, indent=2)
+
+        print(f"State saved to {RESULT_PATH}")
+
+        # From here on you could run things on a different thread/process
+
+        # Reading state from file (demonstrating deserialization)
+        print(f"Loading state from {RESULT_PATH}")
+        with RESULT_PATH.open() as f:
+            stored_state_json = json.load(f)
+
+        state = await RunState.from_json(agent, stored_state_json)
+
+        # Process each interruption
+        for interruption in result.interruptions:
+            print("\nTool call details:")
+            print(f"  Agent: {interruption.agent.name}")
+            print(f"  Tool: {interruption.name}")
+            print(f"  Arguments: {interruption.arguments}")
+
+            confirmed = await confirm("\nDo you approve this tool call?")
+
+            if confirmed:
+                print(f"✓ Approved: {interruption.name}")
+                state.approve(interruption)
+            else:
+                print(f"✗ Rejected: {interruption.name}")
+                state.reject(interruption)
+
+        # Resume execution with the updated state
+        print("\nResuming agent execution...")
+        result = await Runner.run(agent, state)
+        has_interruptions = len(result.interruptions) > 0
+
+    print("\n" + "=" * 80)
+    print("Final Output:")
+    print("=" * 80)
+    print(result.final_output)
+
+
+if __name__ == "__main__":
+    asyncio.run(main())

examples/agent_patterns/human_in_the_loop_stream.py

@@ -0,0 +1,119 @@
+"""Human-in-the-loop example with streaming.
+
+This example demonstrates the human-in-the-loop (HITL) pattern with streaming.
+The agent will pause execution when a tool requiring approval is called,
+allowing you to approve or reject the tool call before continuing.
+
+The streaming version provides real-time feedback as the agent processes
+the request, then pauses for approval when needed.
+"""
+
+import asyncio
+
+from agents import Agent, Runner, function_tool
+from examples.auto_mode import confirm_with_fallback
+
+
+async def _needs_temperature_approval(_ctx, params, _call_id) -> bool:
+    """Check if temperature tool needs approval."""
+    return "Oakland" in params.get("city", "")
+
+
+@function_tool(
+    # Dynamic approval: only require approval for Oakland
+    needs_approval=_needs_temperature_approval
+)
+async def get_temperature(city: str) -> str:
+    """Get the temperature for a given city.
+
+    Args:
+        city: The city to get temperature for.
+
+    Returns:
+        Temperature information for the city.
+    """
+    return f"The temperature in {city} is 20° Celsius"
+
+
+@function_tool
+async def get_weather(city: str) -> str:
+    """Get the weather for a given city.
+
+    Args:
+        city: The city to get weather for.
+
+    Returns:
+        Weather information for the city.
+    """
+    return f"The weather in {city} is sunny."
+
+
+async def confirm(question: str) -> bool:
+    """Prompt user for yes/no confirmation.
+
+    Args:
+        question: The question to ask.
+
+    Returns:
+        True if user confirms, False otherwise.
+    """
+    return confirm_with_fallback(f"{question} (y/n): ", default=True)
+
+
+async def main():
+    """Run the human-in-the-loop example."""
+    main_agent = Agent(
+        name="Weather Assistant",
+        instructions=(
+            "You are a helpful weather assistant. "
+            "Answer questions about weather and temperature using the available tools."
+        ),
+        tools=[get_temperature, get_weather],
+    )
+
+    # Run the agent with streaming
+    result = Runner.run_streamed(
+        main_agent,
+        "What is the weather and temperature in Oakland?",
+    )
+    async for _ in result.stream_events():
+        pass  # Process streaming events silently or could print them
+
+    # Handle interruptions
+    while len(result.interruptions) > 0:
+        print("\n" + "=" * 80)
+        print("Human-in-the-loop: approval required for the following tool calls:")
+        print("=" * 80)
+
+        state = result.to_state()
+
+        for interruption in result.interruptions:
+            print("\nTool call details:")
+            print(f"  Agent: {interruption.agent.name}")
+            print(f"  Tool: {interruption.name}")
+            print(f"  Arguments: {interruption.arguments}")
+
+            confirmed = await confirm("\nDo you approve this tool call?")
+
+            if confirmed:
+                print(f"✓ Approved: {interruption.name}")
+                state.approve(interruption)
+            else:
+                print(f"✗ Rejected: {interruption.name}")
+                state.reject(interruption)
+
+        # Resume execution with streaming
+        print("\nResuming agent execution...")
+        result = Runner.run_streamed(main_agent, state)
+        async for _ in result.stream_events():
+            pass  # Process streaming events silently or could print them
+
+    print("\n" + "=" * 80)
+    print("Final Output:")
+    print("=" * 80)
+    print(result.final_output)
+    print("\nDone!")
+
+
+if __name__ == "__main__":
+    asyncio.run(main())