Early Stopping in ADK SequentialAgent?

[zaferkilic2024]

Is there a built-in mechanism to stop the entire agent sequence in SequentialAgent if a sub-agent meets a certain condition? Or do we need to manually add state checks to each sub-agent?

[dexhunter]

For a condition in your own workflow, put the check in one custom orchestration agent. That avoids adding the same check to every child.

SequentialAgent runs its children in order and does not expose a general stop_condition parameter. A minimal equivalent with a condition between steps is:

from contextlib import aclosing
from google.adk.agents import BaseAgent

class ConditionalSequence(BaseAgent):
    async def _run_async_impl(self, ctx):
        for agent in self.sub_agents:
            if ctx.session.state.get("stop_sequence", False):
                return
            async with aclosing(agent.run_async(ctx)) as events:
                async for event in events:
                    yield event

root_agent = ConditionalSequence(
    name="pipeline",
    sub_agents=[first_agent, second_agent, third_agent],
)

When a child's tool determines that the sequence should stop, set tool_context.state["stop_sequence"] = True. Once that child finishes, the orchestrator skips the remaining children. Reset the flag when starting a new sequence if you reuse the session.

This stops between children, rather than cancelling a child that is already running. Also, actions.escalate is not a general sequential stop flag: the SequentialAgent loop does not use it as a condition to skip subsequent children.