feat: add cascaded interrupt handling for sub-agent tools

[thiagopo]

Description

Motivation: A few days ago, I tried to use interrupts inside a hook triggered by a tool that belongs to a sub-agent. This sub-agent itself was registered as a tool in an orchestrator agent.

Although I managed to make it work, the solution required manually storing additional state, which makes the implementation more complex and less intuitive.

I reproduced this workaround here: https://github.com/thiagopo/strands-agents-interrupt-on-agent-as-tool

What was implemented (resume):

• tool_context.cascade_interrupts(...): agent._interrupt_state.context will be used to store information about cascaded interrupts
• tool_context.get_cascaded_interrupt_responses(): will read from agent._interrupt_state.context and return a list of InterruptResponseContent
• A new internal exception type, e.g. CascadedInterruptException

Key Objectives

• Backward compatibility: The solution must not impact existing interrupt flows (e.g., Swarm, Graph, and current resume behavior)
• Reuse existing mechanisms: Avoid introducing parallel interrupt systems
• Developer experience: The solution should be simple and intuitive to use from the user perspective

Below is an example of how this could be exposed for developers:

class ApprovalHook(HookProvider):

    def register_hooks(self, registry, **kwargs):
        registry.add_callback(BeforeToolCallEvent, self.my_hook)
		
	def my_hook(self, event: BeforeToolCallEvent):
		
		approval = event.interrupt("name_of_my_interrupt", reason="some details...." )
        if approval.lower() not in ["approved", "yes", "ok"]:
            event.cancel_tool = "User canceled the tool execution"


@tool(context=True)
def run_sub_agent(query: str, tool_context: ToolContext) -> str:
    
	sub_agent = agent(...)  # hooks=[ApprovalHook()]

    # Check if resuming from a cascaded interrupt
    responses = tool_context.get_cascaded_interrupt_responses()   <<<<< my proposition
    if responses:
        result = sub_agent(responses)
    else:
        result = sub_agent(query)

    if result.stop_reason == "interrupt":
        # Cascade sub-agent interrupts to orchestrator
        tool_context.cascade_interrupts(result.interrupts)        <<<<<< my proposition (rise a new exception type: CascadedInterruptException)

    return str(result)

Related Issues

#1952 and #1618

Documentation PR

703

Type of Change

New feature

Testing

How have you tested the change?

Yes, I have tested using the example teachers assistants (exaple chagend here: https://github.com/thiagopo/strands-agents-interrupt-on-agent-as-tool/tree/version_for_testing_cascade_interrupts)

• I ran hatch run prepare

Checklist

• I have read the CONTRIBUTING document
• I have added any necessary tests that prove my fix is effective or my feature works
• I have updated the documentation accordingly
• I have added an appropriate example to the documentation to outline the feature, or no new docs are needed
• My changes generate no new warnings
• Any dependent changes have been merged and published

---

By submitting this pull request, I confirm that you can use, modify, copy, and redistribute this contribution, under the terms of your choice.