interrupts - swarm

[pgrayy]

Description

Adding interrupt (HIL) support to Swarm. Will add support to Graph in a separate PR.

Usage

Note, these are the same interfaces available for single agent interrupts (docs).

BeforeNodeCallEvent Hook

import json from strands import Agent from strands.experimental.hooks.multiagent import BeforeNodeCallEvent from strands.hooks import HookProvider from strands.interrupt import Interrupt from strands.multiagent import Swarm from strands.multiagent.base import Status class ApprovalHook(HookProvider): def register_hooks(self, registry): registry.add_callback(BeforeNodeCallEvent, self.approve) def approve(self, event): if event.node_id == "info": return response = event.interrupt("my_interrupt", reason=f"{event.node_id} needs approval") if response != "APPROVE": event.cancel_node = "node rejected" info_agent = Agent(name="info") weather_agent = Agent(name="weather") swarm = Swarm([info_agent, weather_agent], hooks=[ApprovalHook()]) multiagent_result = swarm("What is the weather?") while multiagent_result.status != Status.INTERRUPTED: response = [] for interrupt in multiagent_result.interrupts: if interrupt.name == "my_interrupt": response = input(f"{interrupt.reason} (y/N): ") responses.append( { "interruptResponse": { "interruptId": interrupt.id, "response": response, }, }, ) multiagent_result = swarm(responses) weather_result = multiagent_result.results["weather"] weather_message = json.dumps(weather_result.result.message).lower() print(weather_message)

Within Agent

import json from strands import Agent, tool from strands.interrupt import Interrupt from strands.multiagent import Swarm from strands.multiagent.base import Status @tool(context=True) def user_name_tool(tool_context: ToolContext) -> str: response = tool_context.interrupt("my_interrupt", reason="need user name") return response user_agent = Agent(name="user", tools=[user_name_tool]) swarm = Swarm([user_agent]) multiagent_result = swarm("Who is the current user?") while multiagent_result.status != Status.INTERRUPTED: response = [] for interrupt in multiagent_result.interrupts: if interrupt.name == "my_interrupt": response = input(f"{interrupt.reason}: ") responses.append( { "interruptResponse": { "interruptId": interrupt.id, "response": response, }, }, ) multiagent_result = swarm(responses) user_result = multiagent_result.results["user"] user_message = json.dumps(user_result.result.message).lower() print(user_message)

Related Issues

#204

Documentation PR

After closing out PR, will add usage instructions to https://strandsagents.com/latest/documentation/docs/user-guide/concepts/interrupts/

Type of Change

New feature

Testing

How have you tested the change? Verify that the changes do not break functionality or introduce warnings in consuming repositories: agents-docs, agents-tools, agents-cli

• I ran hatch run prepare: Wrote new unit tests
• I ran hatch test tests_integ/interrupts/multiagent/*.py: Wrote new integ tests

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.

[codecov]

Codecov Report

❌ Patch coverage is 80.68182% with 17 lines in your changes missing coverage. Please review.

Files with missing lines	Patch %	Lines
src/strands/multiagent/swarm.py	83.05%	5 Missing and 5 partials ⚠️
src/strands/multiagent/base.py	40.00%	6 Missing ⚠️
src/strands/multiagent/graph.py	50.00%	1 Missing ⚠️

📢 Thoughts on this report? Let us know!

[pgrayy]

This is for if the node agent itself interrupts. The user can raise an interrupt at the swarm level in the BeforeNodeCallEvent hook. They can also raise interrupts in the agent nodes themselves. If raising an interrupt in an agent, we need to restore its interrupt state on resume, which is what we handle here.

[pgrayy]

Shoot, need to add a check in Graph to temporarily error on use of interrupts. This PR just adds the functionality to Swarm. Graph will be a fast follow. Still, please feel free to review the rest of the PR in the meanwhile.

[pgrayy]

Still need to do this. Apologies for the delay. But still please feel free to review.