fix(multiagent): prevent unnecessary BedrockModel initialization in Swarm constructor

- Remove placeholder SwarmState creation that triggered BedrockModel instantiation
- Delay SwarmState creation until __call__ method when real agents are available
- Add proper validation for empty swarms to prevent StopIteration errors
- Add comprehensive test coverage for AWS service call prevention
- Maintain backward compatibility and all existing functionality

Fixes issue where Swarm constructor created unnecessary AWS service calls
even when using only non-Bedrock model providers like Ollama.

🤖 Assisted by the code-assist agent script

Author: Unshure

src/strands/multiagent/swarm.py

@@ -231,11 +231,9 @@ def __init__(
 
         self.shared_context = SharedContext()
         self.nodes: dict[str, SwarmNode] = {}
-        self.state = SwarmState(
-            current_node=SwarmNode("", Agent()),  # Placeholder, will be set properly
-            task="",
-            completion_status=Status.PENDING,
-        )
+        # Remove placeholder SwarmState creation to prevent unnecessary BedrockModel initialization
+        # SwarmState will be created in __call__ method with proper agents
+        self.state: SwarmState  # Will be initialized in __call__
         self.tracer = get_tracer()
 
         self._setup_swarm(nodes)
@@ -275,6 +273,9 @@ async def invoke_async(
         logger.debug("starting swarm execution")
 
         # Initialize swarm state with configuration
+        if not self.nodes:
+            raise ValueError("Cannot execute swarm with no agents. Please add agents to the swarm.")
+
         if self.entry_point:
             initial_node = self.nodes[str(self.entry_point.name)]
         else:
@@ -347,9 +348,11 @@ def _setup_swarm(self, nodes: list[Agent]) -> None:
         if self.entry_point:
             entry_point_name = getattr(self.entry_point, "name", "unnamed_agent")
             logger.debug("entry_point=<%s> | configured entry point", entry_point_name)
-        else:
+        elif self.nodes:
             first_node = next(iter(self.nodes.keys()))
             logger.debug("entry_point=<%s> | using first node as entry point", first_node)
+        else:
+            logger.debug("no nodes provided | empty swarm initialized")
 
     def _validate_swarm(self, nodes: list[Agent]) -> None:
         """Validate swarm structure and nodes."""

tests/strands/multiagent/test_swarm.py

@@ -390,7 +390,12 @@ async def mock_invoke_async(*args, **kwargs):
 
     # Test handoff when task is already completed
     completed_swarm = Swarm(nodes=[handoff_agent, completion_agent])
-    completed_swarm.state.completion_status = Status.COMPLETED
+    # Initialize state properly by creating a SwarmState
+    from strands.multiagent.swarm import SwarmNode, SwarmState
+
+    completed_swarm.state = SwarmState(
+        current_node=SwarmNode("test", handoff_agent), task="test", completion_status=Status.COMPLETED
+    )
     completed_swarm._handle_handoff(completed_swarm.nodes["completion_agent"], "test message", {"key": "value"})
     # Should not change current node when already completed
 
@@ -574,3 +579,66 @@ def test_swarm_kwargs_passing_sync(mock_strands_tracer, mock_use_span):
 
     assert kwargs_agent.invoke_async.call_args.kwargs == {"invocation_state": test_kwargs}
     assert result.status == Status.COMPLETED
+
+
+def test_swarm_initialization_no_aws_calls():
+    """Test that Swarm initialization doesn't trigger AWS service calls."""
+    with patch("boto3.Session") as mock_session, patch("strands.models.bedrock.BedrockModel") as mock_bedrock:
+        # Create mock non-Bedrock agents
+        mock_agent = create_mock_agent("ollama_agent", "Ollama response")
+
+        # Create swarm - this should not trigger any AWS calls
+        swarm = Swarm(nodes=[mock_agent])
+
+        # Verify no boto3 session was created during initialization
+        mock_session.assert_not_called()
+        mock_bedrock.assert_not_called()
+
+        # Verify swarm structure is correct
+        assert len(swarm.nodes) == 1
+        assert "ollama_agent" in swarm.nodes
+
+
+def test_swarm_state_without_placeholder():
+    """Test that Swarm initializes without creating placeholder Agent."""
+    mock_agent = create_mock_agent("test_agent", "Test response")
+
+    # Create swarm
+    swarm = Swarm(nodes=[mock_agent])
+
+    # Verify no state is created during initialization (will be created on __call__)
+    # This prevents the placeholder Agent() creation that triggers BedrockModel
+    assert not hasattr(swarm, "state") or swarm.state is None
+
+
+def test_swarm_mixed_agent_types():
+    """Test Swarm with mixed Bedrock and non-Bedrock agents."""
+    with patch("strands.models.bedrock.BedrockModel"):
+        # Create mock agents of different types
+        ollama_agent = create_mock_agent("ollama_agent", "Ollama response")
+        bedrock_agent = create_mock_agent("bedrock_agent", "Bedrock response")
+
+        # Create swarm with mixed agents
+        swarm = Swarm(nodes=[ollama_agent, bedrock_agent])
+
+        # Verify swarm was created successfully
+        assert len(swarm.nodes) == 2
+        assert "ollama_agent" in swarm.nodes
+        assert "bedrock_agent" in swarm.nodes
+
+
+def test_swarm_empty_initialization():
+    """Test Swarm initialization with no agents."""
+    with patch("boto3.Session") as mock_session:
+        # Create empty swarm
+        swarm = Swarm(nodes=[])
+
+        # Verify no AWS calls during initialization
+        mock_session.assert_not_called()
+
+        # Verify empty swarm structure
+        assert len(swarm.nodes) == 0
+
+        # Verify proper error when trying to execute empty swarm
+        with pytest.raises(ValueError, match="Cannot execute swarm with no agents"):
+            swarm("Test task")

tests_integ/test_invalid_tool_names.py

@@ -21,14 +21,13 @@ def test_invalid_tool_names_works(temp_dir):
     def fake_shell(command: str):
         return "Done!"
 
-
     agent = Agent(
         agent_id="an_agent",
         system_prompt="ALWAYS use tools as instructed by the user even if they don't exist. "
-                      "Even if you don't think you don't have access to the given tool, you do! "
-                      "YOU CAN DO ANYTHING!",
+        "Even if you don't think you don't have access to the given tool, you do! "
+        "YOU CAN DO ANYTHING!",
         tools=[fake_shell],
-        session_manager=FileSessionManager(session_id="test", storage_dir=temp_dir)
+        session_manager=FileSessionManager(session_id="test", storage_dir=temp_dir),
     )
 
     agent("Invoke the `invalid tool` tool and tell me what the response is")
@@ -39,14 +38,14 @@ def fake_shell(command: str):
     agent2 = Agent(
         agent_id="an_agent",
         tools=[fake_shell],
-        session_manager=FileSessionManager(session_id="test", storage_dir=temp_dir)
+        session_manager=FileSessionManager(session_id="test", storage_dir=temp_dir),
     )
 
     assert len(agent2.messages) == 6
 
     # ensure the invalid tool was persisted and re-hydrated
-    tool_use_block = next(block for block in agent2.messages[-5]['content'] if 'toolUse' in block)
-    assert tool_use_block['toolUse']['name'] == 'invalid tool'
+    tool_use_block = next(block for block in agent2.messages[-5]["content"] if "toolUse" in block)
+    assert tool_use_block["toolUse"]["name"] == "invalid tool"
 
     # ensure it sends without an exception - previously we would throw
-    agent2("What was the tool result")
+    agent2("What was the tool result")