feat: implement human-in-the-loop (HITL) approval infrastructure

This commit adds the foundational components for human-in-the-loop
functionality in the Python OpenAI Agents SDK, matching the TypeScript
implementation.

**Completed Components:**

1. **Tool Approval Field** (tool.py)
   - Added `needs_approval` field to FunctionTool
   - Supports boolean or callable (dynamic approval)
   - Updated function_tool() decorator

2. **ToolApprovalItem Class** (items.py)
   - New item type for tool calls requiring approval
   - Added to RunItem union type

3. **Approval Tracking** (run_context.py)
   - Created ApprovalRecord class
   - Added approval infrastructure to RunContextWrapper
   - Methods: is_tool_approved(), approve_tool(), reject_tool()
   - Supports individual and permanent approvals/rejections

4. **RunState Class** (run_state.py) - NEW FILE
   - Complete serialization/deserialization support
   - approve() and reject() methods
   - get_interruptions() method
   - Agent map building for name resolution
   - 567 lines of serialization logic

5. **Interruptions Support** (result.py)
   - Added interruptions field to RunResultBase
   - Will contain ToolApprovalItem instances when paused

6. **NextStepInterruption** (run_state.py)
   - New step type for representing interruptions

**Remaining Work:**

1. Add NextStepInterruption to NextStep union in _run_impl.py
2. Implement tool approval checking in run execution
3. Update run methods to accept RunState
4. Add comprehensive tests
5. Update documentation

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-Authored-By: Claude <[email protected]>

Author: mjschock

src/agents/__init__.py

@@ -55,6 +55,7 @@
     ModelResponse,
     ReasoningItem,
     RunItem,
+    ToolApprovalItem,
     ToolCallItem,
     ToolCallOutputItem,
     TResponseInputItem,
@@ -77,6 +78,7 @@
 from .result import RunResult, RunResultStreaming
 from .run import RunConfig, Runner
 from .run_context import RunContextWrapper, TContext
+from .run_state import NextStepInterruption, RunState
 from .stream_events import (
     AgentUpdatedStreamEvent,
     RawResponsesStreamEvent,

src/agents/items.py

@@ -327,6 +327,21 @@ class MCPApprovalResponseItem(RunItemBase[McpApprovalResponse]):
     type: Literal["mcp_approval_response_item"] = "mcp_approval_response_item"
 
 
+@dataclass
+class ToolApprovalItem(RunItemBase[ResponseFunctionToolCall]):
+    """Represents a tool call that requires approval before execution.
+
+    When a tool has `needs_approval=True`, the run will be interrupted and this item will be
+    added to the interruptions list. You can then approve or reject the tool call using
+    RunState.approve() or RunState.reject() and resume the run.
+    """
+
+    raw_item: ResponseFunctionToolCall
+    """The raw function tool call that requires approval."""
+
+    type: Literal["tool_approval_item"] = "tool_approval_item"
+
+
 RunItem: TypeAlias = Union[
     MessageOutputItem,
     HandoffCallItem,
@@ -337,6 +352,7 @@ class MCPApprovalResponseItem(RunItemBase[McpApprovalResponse]):
     MCPListToolsItem,
     MCPApprovalRequestItem,
     MCPApprovalResponseItem,
+    ToolApprovalItem,
 ]
 """An item generated by an agent."""
 

src/agents/result.py

@@ -70,6 +70,11 @@ class RunResultBase(abc.ABC):
     context_wrapper: RunContextWrapper[Any]
     """The context wrapper for the agent run."""
 
+    interruptions: list[RunItem]
+    """Any interruptions (e.g., tool approval requests) that occurred during the run.
+    If non-empty, the run was paused waiting for user action (e.g., approve/reject tool calls).
+    """
+
     @property
     @abc.abstractmethod
     def last_agent(self) -> Agent[Any]:

src/agents/run_context.py

@@ -1,13 +1,32 @@
+from __future__ import annotations
+
 from dataclasses import dataclass, field
-from typing import Any, Generic
+from typing import TYPE_CHECKING, Any, Generic
 
 from typing_extensions import TypeVar
 
 from .usage import Usage
 
+if TYPE_CHECKING:
+    from .items import ToolApprovalItem
+
 TContext = TypeVar("TContext", default=Any)
 
 
+class ApprovalRecord:
+    """Tracks approval/rejection state for a tool."""
+
+    approved: bool | list[str]
+    """Either True (always approved), False (never approved), or a list of approved call IDs."""
+
+    rejected: bool | list[str]
+    """Either True (always rejected), False (never rejected), or a list of rejected call IDs."""
+
+    def __init__(self):
+        self.approved = []
+        self.rejected = []
+
+
 @dataclass
 class RunContextWrapper(Generic[TContext]):
     """This wraps the context object that you passed to `Runner.run()`. It also contains
@@ -24,3 +43,116 @@ class RunContextWrapper(Generic[TContext]):
     """The usage of the agent run so far. For streamed responses, the usage will be stale until the
     last chunk of the stream is processed.
     """
+
+    _approvals: dict[str, ApprovalRecord] = field(default_factory=dict)
+    """Internal tracking of tool approval/rejection decisions."""
+
+    def is_tool_approved(self, tool_name: str, call_id: str) -> bool | None:
+        """Check if a tool call has been approved.
+
+        Args:
+            tool_name: The name of the tool being called.
+            call_id: The ID of the specific tool call.
+
+        Returns:
+            True if approved, False if rejected, None if not yet decided.
+        """
+        approval_entry = self._approvals.get(tool_name)
+        if not approval_entry:
+            return None
+
+        # Check for permanent approval/rejection
+        if approval_entry.approved is True and approval_entry.rejected is True:
+            # Approval takes precedence
+            return True
+
+        if approval_entry.approved is True:
+            return True
+
+        if approval_entry.rejected is True:
+            return False
+
+        # Check for individual call approval/rejection
+        individual_approval = (
+            call_id in approval_entry.approved
+            if isinstance(approval_entry.approved, list)
+            else False
+        )
+        individual_rejection = (
+            call_id in approval_entry.rejected
+            if isinstance(approval_entry.rejected, list)
+            else False
+        )
+
+        if individual_approval and individual_rejection:
+            # Approval takes precedence
+            return True
+
+        if individual_approval:
+            return True
+
+        if individual_rejection:
+            return False
+
+        return None
+
+    def approve_tool(self, approval_item: ToolApprovalItem, always_approve: bool = False) -> None:
+        """Approve a tool call.
+
+        Args:
+            approval_item: The tool approval item to approve.
+            always_approve: If True, always approve this tool (for all future calls).
+        """
+        tool_name = approval_item.raw_item.name
+        call_id = approval_item.raw_item.call_id
+
+        if always_approve:
+            approval_entry = ApprovalRecord()
+            approval_entry.approved = True
+            approval_entry.rejected = []
+            self._approvals[tool_name] = approval_entry
+            return
+
+        if tool_name not in self._approvals:
+            self._approvals[tool_name] = ApprovalRecord()
+
+        approval_entry = self._approvals[tool_name]
+        if isinstance(approval_entry.approved, list):
+            approval_entry.approved.append(call_id)
+
+    def reject_tool(self, approval_item: ToolApprovalItem, always_reject: bool = False) -> None:
+        """Reject a tool call.
+
+        Args:
+            approval_item: The tool approval item to reject.
+            always_reject: If True, always reject this tool (for all future calls).
+        """
+        tool_name = approval_item.raw_item.name
+        call_id = approval_item.raw_item.call_id
+
+        if always_reject:
+            approval_entry = ApprovalRecord()
+            approval_entry.approved = False
+            approval_entry.rejected = True
+            self._approvals[tool_name] = approval_entry
+            return
+
+        if tool_name not in self._approvals:
+            self._approvals[tool_name] = ApprovalRecord()
+
+        approval_entry = self._approvals[tool_name]
+        if isinstance(approval_entry.rejected, list):
+            approval_entry.rejected.append(call_id)
+
+    def _rebuild_approvals(self, approvals: dict[str, dict[str, Any]]) -> None:
+        """Rebuild approvals from serialized state (for RunState deserialization).
+
+        Args:
+            approvals: Dictionary mapping tool names to approval records.
+        """
+        self._approvals = {}
+        for tool_name, record_dict in approvals.items():
+            record = ApprovalRecord()
+            record.approved = record_dict.get("approved", [])
+            record.rejected = record_dict.get("rejected", [])
+            self._approvals[tool_name] = record