feat: Skip running a workflow agent if it has no sub-agents

PiperOrigin-RevId: 811528166

src/google/adk/agents/loop_agent.py

@@ -66,6 +66,9 @@ class LoopAgent(BaseAgent):
   async def _run_async_impl(
       self, ctx: InvocationContext
   ) -> AsyncGenerator[Event, None]:
+    if not self.sub_agents:
+      return
+
     times_looped = 0
     while not self.max_iterations or times_looped < self.max_iterations:
       for sub_agent in self.sub_agents:

src/google/adk/agents/parallel_agent.py

@@ -175,6 +175,9 @@ class ParallelAgent(BaseAgent):
   async def _run_async_impl(
       self, ctx: InvocationContext
   ) -> AsyncGenerator[Event, None]:
+    if not self.sub_agents:
+      return
+
     agent_runs = [
         sub_agent.run_async(
             _create_branch_ctx_for_sub_agent(self, sub_agent, ctx)

src/google/adk/agents/sequential_agent.py

@@ -51,6 +51,10 @@ class SequentialAgent(BaseAgent):
   async def _run_async_impl(
       self, ctx: InvocationContext
   ) -> AsyncGenerator[Event, None]:
+    # Skip if there is no sub-agent.
+    if not self.sub_agents:
+      return
+
     for sub_agent in self.sub_agents:
       pause_invocation = False
 
@@ -80,6 +84,9 @@ class SequentialAgent(BaseAgent):
     Args:
       ctx: The invocation context of the agent.
     """
+    if not self.sub_agents:
+      return
+
     # There is no way to know if it's using live during init phase so we have to init it here
     for sub_agent in self.sub_agents:
       # add tool

tests/unittests/agents/test_loop_agent.py

@@ -114,6 +114,20 @@ async def test_run_async(request: pytest.FixtureRequest):
   assert events[1].content.parts[0].text == f'Hello, async {agent.name}!'
 
 
+@pytest.mark.asyncio
+async def test_run_async_skip_if_no_sub_agent(request: pytest.FixtureRequest):
+  loop_agent = LoopAgent(
+      name=f'{request.function.__name__}_test_loop_agent',
+      max_iterations=2,
+      sub_agents=[],
+  )
+  parent_ctx = await _create_parent_invocation_context(
+      request.function.__name__, loop_agent
+  )
+  events = [e async for e in loop_agent.run_async(parent_ctx)]
+  assert not events
+
+
 @pytest.mark.asyncio
 async def test_run_async_with_escalate_action(request: pytest.FixtureRequest):
   non_escalating_agent = _TestingAgent(

tests/unittests/agents/test_parallel_agent.py

@@ -184,6 +184,19 @@ async def test_generating_one_event_per_agent_at_once(
     # Asserts on event are done in _TestingAgentWithMultipleEvents.
 
 
+@pytest.mark.asyncio
+async def test_run_async_skip_if_no_sub_agent(request: pytest.FixtureRequest):
+  parallel_agent = ParallelAgent(
+      name=f'{request.function.__name__}_test_parallel_agent',
+      sub_agents=[],
+  )
+  parent_ctx = await _create_parent_invocation_context(
+      request.function.__name__, parallel_agent
+  )
+  events = [e async for e in parallel_agent.run_async(parent_ctx)]
+  assert not events
+
+
 class _TestingAgentWithException(_TestingAgent):
   """Mock agent for testing."""
 

tests/unittests/agents/test_sequential_agent.py

@@ -91,6 +91,20 @@ async def test_run_async(request: pytest.FixtureRequest):
   assert events[1].content.parts[0].text == f'Hello, async {agent_2.name}!'
 
 
+@pytest.mark.asyncio
+async def test_run_async_skip_if_no_sub_agent(request: pytest.FixtureRequest):
+  sequential_agent = SequentialAgent(
+      name=f'{request.function.__name__}_test_agent',
+      sub_agents=[],
+  )
+  parent_ctx = await _create_parent_invocation_context(
+      request.function.__name__, sequential_agent
+  )
+  events = [e async for e in sequential_agent.run_async(parent_ctx)]
+
+  assert not events
+
+
 @pytest.mark.asyncio
 async def test_run_live(request: pytest.FixtureRequest):
   agent_1 = _TestingAgent(name=f'{request.function.__name__}_test_agent_1')