fix(multiagent): raise ValueError for unsupported Graph and Swarm agent features (#472)

Author: awsarron

src/strands/hooks/registry.py

@@ -204,6 +204,20 @@ def invoke_callbacks(self, event: TInvokeEvent) -> TInvokeEvent:
 
         return event
 
+    def has_callbacks(self) -> bool:
+        """Check if the registry has any registered callbacks.
+
+        Returns:
+            True if there are any registered callbacks, False otherwise.
+
+        Example:
+            ```python
+            if registry.has_callbacks():
+                print("Registry has callbacks registered")
+            ```
+        """
+        return bool(self._registered_callbacks)
+
     def get_callbacks_for(self, event: TEvent) -> Generator[HookCallback[TEvent], None, None]:
         """Get callbacks registered for the given event in the appropriate order.
 

src/strands/multiagent/graph.py

@@ -129,6 +129,32 @@ def __eq__(self, other: Any) -> bool:
         return self.node_id == other.node_id
 
 
+def _validate_node_executor(
+    executor: Agent | MultiAgentBase, existing_nodes: dict[str, GraphNode] | None = None
+) -> None:
+    """Validate a node executor for graph compatibility.
+
+    Args:
+        executor: The executor to validate
+        existing_nodes: Optional dict of existing nodes to check for duplicates
+    """
+    # Check for duplicate node instances
+    if existing_nodes:
+        seen_instances = {id(node.executor) for node in existing_nodes.values()}
+        if id(executor) in seen_instances:
+            raise ValueError("Duplicate node instance detected. Each node must have a unique object instance.")
+
+    # Validate Agent-specific constraints
+    if isinstance(executor, Agent):
+        # Check for session persistence
+        if executor._session_manager is not None:
+            raise ValueError("Session persistence is not supported for Graph agents yet.")
+
+        # Check for callbacks
+        if executor.hooks.has_callbacks():
+            raise ValueError("Agent callbacks are not supported for Graph agents yet.")
+
+
 class GraphBuilder:
     """Builder pattern for constructing graphs."""
 
@@ -140,10 +166,7 @@ def __init__(self) -> None:
 
     def add_node(self, executor: Agent | MultiAgentBase, node_id: str | None = None) -> GraphNode:
         """Add an Agent or MultiAgentBase instance as a node to the graph."""
-        # Check for duplicate node instances
-        seen_instances = {id(node.executor) for node in self.nodes.values()}
-        if id(executor) in seen_instances:
-            raise ValueError("Duplicate node instance detected. Each node must have a unique object instance.")
+        _validate_node_executor(executor, self.nodes)
 
         # Auto-generate node_id if not provided
         if node_id is None:
@@ -304,6 +327,9 @@ def _validate_graph(self, nodes: dict[str, GraphNode]) -> None:
                 raise ValueError("Duplicate node instance detected. Each node must have a unique object instance.")
             seen_instances.add(id(node.executor))
 
+            # Validate Agent-specific constraints for each node
+            _validate_node_executor(node.executor)
+
     async def _execute_graph(self) -> None:
         """Unified execution flow with conditional routing."""
         ready_nodes = list(self.entry_points)

src/strands/multiagent/swarm.py

@@ -314,6 +314,14 @@ def _validate_swarm(self, nodes: list[Agent]) -> None:
                 raise ValueError("Duplicate node instance detected. Each node must have a unique object instance.")
             seen_instances.add(id(node))
 
+            # Check for session persistence
+            if node._session_manager is not None:
+                raise ValueError("Session persistence is not supported for Swarm agents yet.")
+
+            # Check for callbacks
+            if node.hooks.has_callbacks():
+                raise ValueError("Agent callbacks are not supported for Swarm agents yet.")
+
     def _inject_swarm_tools(self) -> None:
         """Add swarm coordination tools to each agent."""
         # Create tool functions with proper closures

tests/strands/experimental/hooks/test_hook_registry.py

@@ -150,3 +150,18 @@ def callback2(_event):
     hook_registry.invoke_callbacks(test_after_event)
 
     assert call_order == ["callback2", "callback1"]  # Reverse order
+
+
+def test_has_callbacks(hook_registry, test_event):
+    """Test that has_callbacks returns correct boolean values."""
+    # Empty registry should return False
+    assert not hook_registry.has_callbacks()
+
+    # Registry with callbacks should return True
+    callback = Mock()
+    hook_registry.add_callback(TestEvent, callback)
+    assert hook_registry.has_callbacks()
+
+    # Test with multiple event types
+    hook_registry.add_callback(TestAfterEvent, Mock())
+    assert hook_registry.has_callbacks()

tests/strands/multiagent/test_graph.py

@@ -3,15 +3,20 @@
 import pytest
 
 from strands.agent import Agent, AgentResult
+from strands.hooks import AgentInitializedEvent
+from strands.hooks.registry import HookProvider, HookRegistry
 from strands.multiagent.base import MultiAgentBase, MultiAgentResult, NodeResult
-from strands.multiagent.graph import GraphBuilder, GraphEdge, GraphNode, GraphResult, GraphState, Status
+from strands.multiagent.graph import Graph, GraphBuilder, GraphEdge, GraphNode, GraphResult, GraphState, Status
+from strands.session.session_manager import SessionManager
 
 
 def create_mock_agent(name, response_text="Default response", metrics=None, agent_id=None):
     """Create a mock Agent with specified properties."""
     agent = Mock(spec=Agent)
     agent.name = name
     agent.id = agent_id or f"{name}_id"
+    agent._session_manager = None
+    agent.hooks = HookRegistry()
 
     if metrics is None:
         metrics = Mock(
@@ -261,6 +266,10 @@ async def test_graph_execution_with_failures(mock_strands_tracer, mock_use_span)
     failing_agent.id = "fail_node"
     failing_agent.__call__ = Mock(side_effect=Exception("Simulated failure"))
 
+    # Add required attributes for validation
+    failing_agent._session_manager = None
+    failing_agent.hooks = HookRegistry()
+
     async def mock_invoke_failure(*args, **kwargs):
         raise Exception("Simulated failure")
 
@@ -489,3 +498,51 @@ def test_graph_synchronous_execution(mock_strands_tracer, mock_use_span, mock_ag
 
     mock_strands_tracer.start_multiagent_span.assert_called()
     mock_use_span.assert_called_once()
+
+
+def test_graph_validate_unsupported_features():
+    """Test Graph validation for session persistence and callbacks."""
+    # Test with normal agent (should work)
+    normal_agent = create_mock_agent("normal_agent")
+    normal_agent._session_manager = None
+    normal_agent.hooks = HookRegistry()
+
+    builder = GraphBuilder()
+    builder.add_node(normal_agent)
+    graph = builder.build()
+    assert len(graph.nodes) == 1
+
+    # Test with session manager (should fail in GraphBuilder.add_node)
+    mock_session_manager = Mock(spec=SessionManager)
+    agent_with_session = create_mock_agent("agent_with_session")
+    agent_with_session._session_manager = mock_session_manager
+    agent_with_session.hooks = HookRegistry()
+
+    builder = GraphBuilder()
+    with pytest.raises(ValueError, match="Session persistence is not supported for Graph agents yet"):
+        builder.add_node(agent_with_session)
+
+    # Test with callbacks (should fail in GraphBuilder.add_node)
+    class TestHookProvider(HookProvider):
+        def register_hooks(self, registry, **kwargs):
+            registry.add_callback(AgentInitializedEvent, lambda e: None)
+
+    agent_with_hooks = create_mock_agent("agent_with_hooks")
+    agent_with_hooks._session_manager = None
+    agent_with_hooks.hooks = HookRegistry()
+    agent_with_hooks.hooks.add_hook(TestHookProvider())
+
+    builder = GraphBuilder()
+    with pytest.raises(ValueError, match="Agent callbacks are not supported for Graph agents yet"):
+        builder.add_node(agent_with_hooks)
+
+    # Test validation in Graph constructor (when nodes are passed directly)
+    # Test with session manager in Graph constructor
+    node_with_session = GraphNode("node_with_session", agent_with_session)
+    with pytest.raises(ValueError, match="Session persistence is not supported for Graph agents yet"):
+        Graph(nodes={"node_with_session": node_with_session}, edges=set(), entry_points=set())
+
+    # Test with callbacks in Graph constructor
+    node_with_hooks = GraphNode("node_with_hooks", agent_with_hooks)
+    with pytest.raises(ValueError, match="Agent callbacks are not supported for Graph agents yet"):
+        Graph(nodes={"node_with_hooks": node_with_hooks}, edges=set(), entry_points=set())

tests/strands/multiagent/test_swarm.py

@@ -6,8 +6,11 @@
 
 from strands.agent import Agent, AgentResult
 from strands.agent.state import AgentState
+from strands.hooks import AgentInitializedEvent
+from strands.hooks.registry import HookProvider, HookRegistry
 from strands.multiagent.base import Status
 from strands.multiagent.swarm import SharedContext, Swarm, SwarmNode, SwarmResult, SwarmState
+from strands.session.session_manager import SessionManager
 from strands.types.content import ContentBlock
 
 
@@ -27,6 +30,8 @@ def create_mock_agent(
     agent._complete_after = complete_after_calls
     agent._swarm_ref = None  # Will be set by the swarm
     agent._should_fail = should_fail
+    agent._session_manager = None
+    agent.hooks = HookRegistry()
 
     if metrics is None:
         metrics = Mock(
@@ -450,3 +455,36 @@ def test_swarm_metrics_handling():
 
     result = no_metrics_swarm("Test no metrics")
     assert result.status == Status.COMPLETED
+
+
+def test_swarm_validate_unsupported_features():
+    """Test Swarm validation for session persistence and callbacks."""
+    # Test with normal agent (should work)
+    normal_agent = create_mock_agent("normal_agent")
+    normal_agent._session_manager = None
+    normal_agent.hooks = HookRegistry()
+
+    swarm = Swarm([normal_agent])
+    assert len(swarm.nodes) == 1
+
+    # Test with session manager (should fail)
+    mock_session_manager = Mock(spec=SessionManager)
+    agent_with_session = create_mock_agent("agent_with_session")
+    agent_with_session._session_manager = mock_session_manager
+    agent_with_session.hooks = HookRegistry()
+
+    with pytest.raises(ValueError, match="Session persistence is not supported for Swarm agents yet"):
+        Swarm([agent_with_session])
+
+    # Test with callbacks (should fail)
+    class TestHookProvider(HookProvider):
+        def register_hooks(self, registry, **kwargs):
+            registry.add_callback(AgentInitializedEvent, lambda e: None)
+
+    agent_with_hooks = create_mock_agent("agent_with_hooks")
+    agent_with_hooks._session_manager = None
+    agent_with_hooks.hooks = HookRegistry()
+    agent_with_hooks.hooks.add_hook(TestHookProvider())
+
+    with pytest.raises(ValueError, match="Agent callbacks are not supported for Swarm agents yet"):
+        Swarm([agent_with_hooks])