multiagent

Author: pgrayy

src/strands/agent/agent.py

@@ -541,7 +541,7 @@ async def structured_output_async(self, output_model: Type[T], prompt: AgentInpu
                 if not self.messages and not prompt:
                     raise ValueError("No conversation history or prompt provided")
 
-                temp_messages: Messages = self.messages + self._convert_prompt_to_messages(prompt)
+                temp_messages: Messages = self.messages + await self._convert_prompt_to_messages(prompt)
 
                 structured_output_span.set_attributes(
                     {
@@ -657,7 +657,7 @@ async def stream_async(
             callback_handler = kwargs.get("callback_handler", self.callback_handler)
 
         # Process input and get message to add (if any)
-        messages = self._convert_prompt_to_messages(prompt)
+        messages = await self._convert_prompt_to_messages(prompt)
 
         self.trace_span = self._start_agent_trace_span(messages)
 
@@ -812,7 +812,7 @@ async def _execute_event_loop_cycle(
             if structured_output_context:
                 structured_output_context.cleanup(self.tool_registry)
 
-    def _convert_prompt_to_messages(self, prompt: AgentInput) -> Messages:
+    async def _convert_prompt_to_messages(self, prompt: AgentInput) -> Messages:
         if self._interrupt_state.activated:
             return []
 
@@ -827,7 +827,7 @@ def _convert_prompt_to_messages(self, prompt: AgentInput) -> Messages:
                 tool_use_ids = [
                     content["toolUse"]["toolUseId"] for content in self.messages[-1]["content"] if "toolUse" in content
                 ]
-                self._append_message(
+                await self._append_message(
                     {
                         "role": "user",
                         "content": generate_missing_tool_result_content(tool_use_ids),

src/strands/multiagent/graph.py

@@ -453,7 +453,7 @@ def __init__(
         self._resume_from_session = False
         self.id = id
 
-        self.hooks.invoke_callbacks(MultiAgentInitializedEvent(self))
+        run_async(lambda: self.hooks.invoke_callbacks_async(MultiAgentInitializedEvent(self)))
 
     def __call__(
         self, task: str | list[ContentBlock], invocation_state: dict[str, Any] | None = None, **kwargs: Any
@@ -516,7 +516,7 @@ async def stream_async(
         if invocation_state is None:
             invocation_state = {}
 
-        self.hooks.invoke_callbacks(BeforeMultiAgentInvocationEvent(self, invocation_state))
+        await self.hooks.invoke_callbacks_async(BeforeMultiAgentInvocationEvent(self, invocation_state))
 
         logger.debug("task=<%s> | starting graph execution", task)
 
@@ -569,7 +569,7 @@ async def stream_async(
                 raise
             finally:
                 self.state.execution_time = round((time.time() - start_time) * 1000)
-                self.hooks.invoke_callbacks(AfterMultiAgentInvocationEvent(self))
+                await self.hooks.invoke_callbacks_async(AfterMultiAgentInvocationEvent(self))
                 self._resume_from_session = False
                 self._resume_next_nodes.clear()
 
@@ -776,7 +776,7 @@ def _is_node_ready_with_conditions(self, node: GraphNode, completed_batch: list[
 
     async def _execute_node(self, node: GraphNode, invocation_state: dict[str, Any]) -> AsyncIterator[Any]:
         """Execute a single node and yield TypedEvent objects."""
-        self.hooks.invoke_callbacks(BeforeNodeCallEvent(self, node.node_id, invocation_state))
+        await self.hooks.invoke_callbacks_async(BeforeNodeCallEvent(self, node.node_id, invocation_state))
 
         # Reset the node's state if reset_on_revisit is enabled, and it's being revisited
         if self.reset_on_revisit and node in self.state.completed_nodes:
@@ -920,7 +920,7 @@ async def _execute_node(self, node: GraphNode, invocation_state: dict[str, Any])
             raise
 
         finally:
-            self.hooks.invoke_callbacks(AfterNodeCallEvent(self, node.node_id, invocation_state))
+            await self.hooks.invoke_callbacks_async(AfterNodeCallEvent(self, node.node_id, invocation_state))
 
     def _accumulate_metrics(self, node_result: NodeResult) -> None:
         """Accumulate metrics from a node result."""

src/strands/multiagent/swarm.py

@@ -273,7 +273,7 @@ def __init__(
 
         self._setup_swarm(nodes)
         self._inject_swarm_tools()
-        self.hooks.invoke_callbacks(MultiAgentInitializedEvent(self))
+        run_async(lambda: self.hooks.invoke_callbacks_async(MultiAgentInitializedEvent(self)))
 
     def __call__(
         self, task: str | list[ContentBlock], invocation_state: dict[str, Any] | None = None, **kwargs: Any
@@ -336,7 +336,7 @@ async def stream_async(
         if invocation_state is None:
             invocation_state = {}
 
-        self.hooks.invoke_callbacks(BeforeMultiAgentInvocationEvent(self, invocation_state))
+        await self.hooks.invoke_callbacks_async(BeforeMultiAgentInvocationEvent(self, invocation_state))
 
         logger.debug("starting swarm execution")
 
@@ -375,7 +375,7 @@ async def stream_async(
                 raise
             finally:
                 self.state.execution_time = round((time.time() - self.state.start_time) * 1000)
-                self.hooks.invoke_callbacks(AfterMultiAgentInvocationEvent(self, invocation_state))
+                await self.hooks.invoke_callbacks_async(AfterMultiAgentInvocationEvent(self, invocation_state))
                 self._resume_from_session = False
 
             # Yield final result after execution_time is set
@@ -687,7 +687,9 @@ async def _execute_swarm(self, invocation_state: dict[str, Any]) -> AsyncIterato
                 # TODO: Implement cancellation token to stop _execute_node from continuing
                 try:
                     # Execute with timeout wrapper for async generator streaming
-                    self.hooks.invoke_callbacks(BeforeNodeCallEvent(self, current_node.node_id, invocation_state))
+                    await self.hooks.invoke_callbacks_async(
+                        BeforeNodeCallEvent(self, current_node.node_id, invocation_state)
+                    )
                     node_stream = self._stream_with_timeout(
                         self._execute_node(current_node, self.state.task, invocation_state),
                         self.node_timeout,
@@ -699,7 +701,9 @@ async def _execute_swarm(self, invocation_state: dict[str, Any]) -> AsyncIterato
                     self.state.node_history.append(current_node)
 
                     #  After self.state add current node, swarm state finish updating, we persist here
-                    self.hooks.invoke_callbacks(AfterNodeCallEvent(self, current_node.node_id, invocation_state))
+                    await self.hooks.invoke_callbacks_async(
+                        AfterNodeCallEvent(self, current_node.node_id, invocation_state)
+                    )
 
                     logger.debug("node=<%s> | node execution completed", current_node.node_id)
 

tests/strands/agent/hooks/test_hook_registry.py

@@ -113,15 +113,16 @@ def test_get_callbacks_for_after_event(hook_registry, after_event):
     assert callbacks[1] == callback1  # Reverse order
 
 
-def test_invoke_callbacks(hook_registry, normal_event):
+@pytest.mark.asyncio
+async def test_invoke_callbacks(hook_registry, normal_event):
     """Test that invoke_callbacks calls all registered callbacks for an event."""
     callback1 = Mock()
     callback2 = Mock()
 
     hook_registry.add_callback(NormalTestEvent, callback1)
     hook_registry.add_callback(NormalTestEvent, callback2)
 
-    hook_registry.invoke_callbacks(normal_event)
+    await hook_registry.invoke_callbacks_async(normal_event)
 
     callback1.assert_called_once_with(normal_event)
     callback2.assert_called_once_with(normal_event)
@@ -134,7 +135,8 @@ def test_invoke_callbacks_no_registered_callbacks(hook_registry, normal_event):
     # Test passes if no exception is raised
 
 
-def test_invoke_callbacks_after_event(hook_registry, after_event):
+@pytest.mark.asyncio
+async def test_invoke_callbacks_after_event(hook_registry, after_event):
     """Test that invoke_callbacks calls callbacks in reverse order for after events."""
     call_order: List[str] = []
 
@@ -147,7 +149,7 @@ def callback2(_event):
     hook_registry.add_callback(AfterTestEvent, callback1)
     hook_registry.add_callback(AfterTestEvent, callback2)
 
-    hook_registry.invoke_callbacks(after_event)
+    await hook_registry.invoke_callbacks_async(after_event)
 
     assert call_order == ["callback2", "callback1"]  # Reverse order
 

tests/strands/agent/test_agent.py

@@ -2240,8 +2240,8 @@ def test_agent_backwards_compatibility_single_text_block():
 
     # Should extract text for backwards compatibility
     assert agent.system_prompt == text
-    
-    
+
+
 @pytest.mark.parametrize(
     "content, expected",
     [

tests/strands/event_loop/test_event_loop.py

@@ -1,6 +1,6 @@
 import concurrent
 import unittest.mock
-from unittest.mock import ANY, MagicMock, call, patch
+from unittest.mock import ANY, AsyncMock, MagicMock, call, patch
 
 import pytest
 
@@ -750,6 +750,7 @@ async def test_request_state_initialization(alist):
     # not setting this to False results in endless recursion
     mock_agent._interrupt_state.activated = False
     mock_agent.event_loop_metrics.start_cycle.return_value = (0, MagicMock())
+    mock_agent.hooks.invoke_callbacks_async = AsyncMock()
 
     # Call without providing request_state
     stream = strands.event_loop.event_loop.event_loop_cycle(

tests/strands/event_loop/test_event_loop_structured_output.py

@@ -1,6 +1,6 @@
 """Tests for structured output integration in the event loop."""
 
-from unittest.mock import Mock, patch
+from unittest.mock import AsyncMock, Mock, patch
 
 import pytest
 from pydantic import BaseModel
@@ -38,10 +38,10 @@ def mock_agent():
     agent.tool_registry = ToolRegistry()
     agent.event_loop_metrics = EventLoopMetrics()
     agent.hooks = Mock()
-    agent.hooks.invoke_callbacks = Mock()
+    agent.hooks.invoke_callbacks_async = AsyncMock()
     agent.trace_span = None
     agent.tool_executor = Mock()
-    agent._append_message = Mock()
+    agent._append_message = AsyncMock()
 
     # Set up _interrupt_state properly
     agent._interrupt_state = Mock()

tests/strands/experimental/hooks/test_hook_aliases.py

@@ -9,6 +9,8 @@
 import sys
 from unittest.mock import Mock
 
+import pytest
+
 from strands.experimental.hooks import (
     AfterModelInvocationEvent,
     AfterToolInvocationEvent,
@@ -80,7 +82,8 @@ def test_after_model_call_event_type_equality():
     assert isinstance(after_model_event, AfterModelCallEvent)
 
 
-def test_experimental_aliases_in_hook_registry():
+@pytest.mark.asyncio
+async def test_experimental_aliases_in_hook_registry():
     """Verify that experimental aliases work with hook registry callbacks."""
     hook_registry = HookRegistry()
     callback_called = False
@@ -103,7 +106,7 @@ def experimental_callback(event: BeforeToolInvocationEvent):
     )
 
     # Invoke callbacks - should work since alias points to same type
-    hook_registry.invoke_callbacks(test_event)
+    await hook_registry.invoke_callbacks_async(test_event)
 
     assert callback_called
     assert received_event is test_event

tests/strands/hooks/test_registry.py

@@ -19,7 +19,8 @@ def agent():
     return instance
 
 
-def test_hook_registry_invoke_callbacks_interrupt(registry, agent):
+@pytest.mark.asyncio
+async def test_hook_registry_invoke_callbacks_interrupt(registry, agent):
     event = BeforeToolCallEvent(
         agent=agent,
         selected_tool=None,
@@ -35,7 +36,7 @@ def test_hook_registry_invoke_callbacks_interrupt(registry, agent):
     registry.add_callback(BeforeToolCallEvent, callback2)
     registry.add_callback(BeforeToolCallEvent, callback3)
 
-    _, tru_interrupts = registry.invoke_callbacks(event)
+    _, tru_interrupts = await registry.invoke_callbacks_async(event)
     exp_interrupts = [
         Interrupt(
             id="v1:before_tool_call:test_tool_id:da3551f3-154b-5978-827e-50ac387877ee",
@@ -55,7 +56,8 @@ def test_hook_registry_invoke_callbacks_interrupt(registry, agent):
     callback3.assert_called_once_with(event)
 
 
-def test_hook_registry_invoke_callbacks_interrupt_name_clash(registry, agent):
+@pytest.mark.asyncio
+async def test_hook_registry_invoke_callbacks_interrupt_name_clash(registry, agent):
     event = BeforeToolCallEvent(
         agent=agent,
         selected_tool=None,
@@ -70,4 +72,4 @@ def test_hook_registry_invoke_callbacks_interrupt_name_clash(registry, agent):
     registry.add_callback(BeforeToolCallEvent, callback2)
 
     with pytest.raises(ValueError, match="interrupt_name=<test_name> | interrupt name used more than once"):
-        registry.invoke_callbacks(event)
+        await registry.invoke_callbacks_async(event)