Return of Control

[pgrayy]

Return of Control

Overview

We're designing a "Return of Control" pattern to enable human-in-the-loop workflows in Strands Agents. This will allow agents to pause execution for external approval, input, or review, then resume seamlessly.

The Problem

Currently, once an agent's event loop starts, it runs to completion without interruption. This prevents scenarios like:

• Tool approval workflows ("Should I delete these files?")
• Security reviews for sensitive operations
• Budget approvals for costly actions
• External system integration points

Proposed Solution

The following are ideas we have begun to explore (not final decisions):

1. Pause - Exception-Based Interruption

Hooks can raise special AgentInterruptException types to cleanly interrupt execution:

class ToolApprovalRequiredException(AgentInterruptException):
    def __init__(self, message: str, context: dict = None):
        super().__init__(message)
        self.context = context or {}

class ApprovalHook(HookProvider):
    def check_approval(self, event: AfterModelInvocationEvent):
        if self.needs_approval(event):
            raise ToolApprovalRequiredException(
                "Approval required for sensitive tool",
                context={"tool_name": "delete_file", "input": {...}}
            )

2. Resume - Null Prompt Detection

Resume by calling the agent with no prompt (agent()) - this triggers automatic resume detection:

try:
    result = agent("Delete important files")
except ToolApprovalRequiredException as e:
    approval = get_user_approval(e.context)
    if approval:
        result = agent()  # Empty call resumes execution

3. State Check - Agent Status Property

Track agent state with a simple status property:

# Check current status
print(f"Agent status: {agent.state.status}")  # "ready", "paused", "running"

# Status automatically updates during pause/resume

Complete Example

from strands import Agent
from strands.types.exceptions import ToolApprovalRequiredException

class ApprovalHook(HookProvider):
    def register_hooks(self, registry):
        registry.add_callback(AfterModelInvocationEvent, self.check_approval)
    
    def check_approval(self, event: AfterModelInvocationEvent):
        # Check for sensitive tools and interrupt if needed
        if self.is_sensitive_tool(event):
            raise ToolApprovalRequiredException(
                f"Approval required for {tool_name}",
                context={"tool_name": tool_name, "input": tool_input}
            )

# Usage
agent = Agent(tools=["delete_file"], hooks=[ApprovalHook()])

try:
    result = agent("Delete old log files")
    print(f"✅ Completed: {result.message}")
    
except ToolApprovalRequiredException as e:
    print(f"⏸️ Status: {agent.state.status}")  # "paused"
    
    if get_user_approval(e.context):
        result = agent()  # Resume
        print(f"✅ Completed: {result.message}")
        print(f"📊 Status: {agent.state.status}")  # "ready"

Session Persistence

Paused agents can be resumed across different processes or sessions:

# Session 1: Agent gets paused
session_manager = FileSessionManager("user_123")
agent = Agent(session_manager=session_manager, hooks=[ApprovalHook()])
# ... agent pauses and state is saved

# Session 2: Resume from different process
agent = Agent(session_manager=session_manager, hooks=[ApprovalHook()])
if agent.state.status == "paused":
    result = agent()  # Resume from saved state