How to Build Multi-Agent Systems Using ADK's SequentialAgent, ParallelAgent, and LoopAgent with agents-cli

To build multi-agent systems using ADK's SequentialAgent, ParallelAgent, and LoopAgent with agents-cli, wrap your sub-agents in these deterministic workflow classes, declare output_key for state passing, and wire the composition into an ADK App that the CLI executes.

The google/agents-cli repository provides the command-line tooling to scaffold and run Agent Development Kit (ADK) projects. When you build multi-agent systems using ADK's SequentialAgent, ParallelAgent, and LoopAgent with agents-cli, you create deterministic control flows that orchestrate LLM-driven agents without relying on the model for routing decisions. These three workflow agents—defined in google.adk.agents according to the reference documentation in skills/google-agents-cli-adk-code/references/adk-python.md—enable sequential pipelines, parallel fan-outs, and iterative loops.

Understanding the Three Workflow Agents

Google ADK provides three deterministic workflow agents that subclass BaseAgent. According to the ADK Python cheatsheet found in skills/google-agents-cli-adk-code/references/adk-python.md, these agents manage execution flow explicitly rather than through LLM-generated control logic.

SequentialAgent

SequentialAgent runs its sub_agents one after another, passing state forward through the shared session. This is ideal for simple pipelines such as retrieve → summarize → ask workflows.

# src/my_project/agent.py

from google.adk.agents import SequentialAgent, Agent

summarizer = Agent(
    name="summarizer",
    model="gemini-flash-latest",
    instruction="Summarize the user input.",
    output_key="summary",               # stored in session.state["summary"]

)

question_gen = Agent(
    name="question_generator",
    model="gemini-flash-latest",
    instruction="Generate three questions based on: {summary}",
)

pipeline = SequentialAgent(
    name="pipeline",
    sub_agents=[summarizer, question_gen],
)

The pipeline runs summarizer first, then passes the generated summary to question_gen via the shared session state. As documented in the reference file, the output_key automatically stores results in session.state[output_key] for downstream retrieval.

ParallelAgent

ParallelAgent starts all sub_agents simultaneously; each must write to a distinct output_key. Results are gathered when every branch finishes, making this suitable for fan-out operations like fetching multiple data sources concurrently.

# src/my_project/agent.py

from google.adk.agents import ParallelAgent, Agent, SequentialAgent

fetch_a = Agent(
    name="fetch_a",
    model="gemini-flash-latest",
    instruction="Retrieve data A.",
    output_key="data_a",
)

fetch_b = Agent(
    name="fetch_b",
    model="gemini-flash-latest",
    instruction="Retrieve data B.",
    output_key="data_b",
)

merger = Agent(
    name="merger",
    model="gemini-flash-latest",
    instruction="Combine data_a: {data_a} and data_b: {data_b}",
)

pipeline = SequentialAgent(
    name="full_pipeline",
    sub_agents=[
        ParallelAgent(name="fetchers", sub_agents=[fetch_a, fetch_b]),
        merger,
    ],
)

As implemented in the ADK source, ParallelAgent launches fetch_a and fetch_b concurrently. The merger agent runs only after both parallel branches complete and write their results to the session state.

LoopAgent

LoopAgent repeats a set of sub_agents until a maximum iteration count is reached or an event with escalate=True is emitted. This supports iterative refinement, validation loops, or human-in-the-loop retries.

# src/my_project/agent.py

from google.adk.agents import LoopAgent, Agent

evaluator = Agent(
    name="evaluator",
    model="gemini-flash-latest",
    instruction="Score the draft (pass/fail).",
    output_schema=Evaluation,   # see the Evaluation BaseModel below

)

refiner = Agent(
    name="refiner",
    model="gemini-flash-latest",
    instruction="Improve the draft based on the evaluator feedback.",
)

pipeline = LoopAgent(
    name="refinement_loop",
    sub_agents=[evaluator, refiner],
    max_iterations=5,           # stop after five passes if not escalated

)

The LoopAgent executes evaluator → refiner repeatedly. If a sub-agent emits Event(..., actions=EventActions(escalate=True))—for example, when the evaluator returns "pass"—the loop terminates early, as noted in skills/google-agents-cli-adk-code/references/adk-python.md.

Wiring the Root Agent into agents-cli

The agents-cli tool scaffolds the project structure and wires your composed root agent into the ADK runtime. The implementation expects workflow agents to be registered within an ADK App.

Scaffold a New Project

adk scaffold my_project
cd my_project

This creates the src/my_project/ directory structure with boilerplate agent.py and app.py files.

Define the Application Entry Point

Replace the generated src/my_project/agent.py with one of the composition patterns shown above. Then ensure src/my_project/app.py wires the root agent:

# src/my_project/app.py

from google.adk.apps import App
from .agent import pipeline   # <- your root agent

app = App(name="my_project", root_agent=pipeline)

The App class, imported from google.adk.apps, registers your workflow agent as the entry point for the runtime.

Run Locally

adk run --app_path src/my_project

The CLI starts an in-memory runner, creates a session, and streams events back to the console. The runner internals in src/google/agents/cli/eval/_synthesize_runner.py and src/google/agents/cli/eval/_inference_runner.py handle the execution of these workflow agents, expecting the tools attribute when processing agents like SequentialAgent.

How State Flows Between Agents

Understanding state management is critical when you build multi-agent systems using ADK's SequentialAgent, ParallelAgent, and LoopAgent with agents-cli.

• output_key – Each agent can declare an output_key. When the agent finishes, the value is automatically stored in session.state[output_key]. Downstream agents retrieve it via {output_key} placeholders in their instruction strings.
• ctx.state – For custom BaseAgent implementations (non-LLM agents), you can read from and write to the dictionary directly.
• Event-based termination – In LoopAgent, a sub-agent may return an Event with actions=EventActions(escalate=True) to break the loop early. This is distinct from the conditional routing used in the graph-based Workflow API.

Testing and Debugging

The repository includes test utilities such as InMemoryRunner, App, and SessionService. A minimal test for the Sequential pipeline looks like:

import pytest
from google.adk.runners import InMemoryRunner
from google.adk.apps import App
from google.genai import types

@pytest.mark.asyncio
async def test_sequential():
    # assume `pipeline` defined as above

    app = App(name="my_project", root_agent=pipeline)
    runner = InMemoryRunner(app=app)

    session = await runner.session_service.create_session(app_name="my_project", user_id="u1")
    async for ev in runner.run_async(
        user_id="u1",
        session_id=session.id,
        new_message=types.Content(role="user", parts=[types.Part.from_text("Explain ADK")]),
    ):
        if ev.output:
            assert isinstance(ev.output, str)

Running pytest ensures the composed agents execute without runtime errors before deployment.

Complete Multi-Agent System Example

Below is a complete minimal project that demonstrates all three agents working together:

# src/multi_agent/agent.py

from google.adk.agents import SequentialAgent, ParallelAgent, LoopAgent, Agent
from pydantic import BaseModel, Field

# ---- Simple agents -------------------------------------------------

summarizer = Agent(
    name="summarizer",
    model="gemini-flash-latest",
    instruction="Summarize the user request.",
    output_key="summary",
)

question_gen = Agent(
    name="question_generator",
    model="gemini-flash-latest",
    instruction="Create three follow-up questions from: {summary}",
)

fetch_a = Agent(
    name="fetch_a",
    model="gemini-flash-latest",
    instruction="Get the latest news about AI.",
    output_key="news_a",
)

fetch_b = Agent(
    name="fetch_b",
    model="gemini-flash-latest",
    instruction="Get the latest research paper titles about AI.",
    output_key="news_b",
)

merger = Agent(
    name="merger",
    model="gemini-flash-latest",
    instruction="Combine news_a and news_b into a short briefing.",
)

class EvalResult(BaseModel):
    grade: str = Field(description="pass or fail")
    comment: str

evaluator = Agent(
    name="evaluator",
    model="gemini-flash-latest",
    instruction="Score the briefing. Return pass/fail.",
    output_schema=EvalResult,
    output_key="eval",
)

refiner = Agent(
    name="refiner",
    model="gemini-flash-latest",
    instruction="Improve the briefing based on {eval.comment}.",
)

# ---- Composition --------------------------------------------------

parallel_fetch = ParallelAgent(name="parallel_fetch", sub_agents=[fetch_a, fetch_b])

pipeline = SequentialAgent(
    name="pipeline",
    sub_agents=[
        summarizer,
        question_gen,
        parallel_fetch,
        merger,
        LoopAgent(
            name="refinement_loop",
            sub_agents=[evaluator, refiner],
            max_iterations=3,
        ),
    ],
)

# src/multi_agent/app.py

from google.adk.apps import App
from .agent import pipeline

app = App(name="multi_agent", root_agent=pipeline)

Execute with:

adk run --app_path src/multi_agent

This configuration runs a sequential flow, performs a parallel fetch, and executes an iterative refinement loop—all orchestrated deterministically without LLM-generated control logic.

Summary

• SequentialAgent executes sub-agents linearly, passing state via output_key through the shared session.
• ParallelAgent runs sub-agents concurrently, requiring distinct output_key values for each branch to prevent state collisions.
• LoopAgent repeats sub-agents until max_iterations is reached or an escalate=True event is emitted, enabling iterative refinement.
• The google.adk.agents module defines these classes, documented in skills/google-agents-cli-adk-code/references/adk-python.md and skills/google-agents-cli-adk-code/SKILL.md.
• Use adk scaffold to create projects, define your composition in src/my_project/agent.py, wire the root agent in app.py, and run with adk run.
• State flows automatically through session.state when using output_key, or directly via ctx.state for custom agents.

Frequently Asked Questions

How do I pass data between agents in a SequentialAgent pipeline?

Use the output_key parameter when defining an agent. When the agent completes, its result is stored in session.state[output_key]. Downstream agents reference this value in their instruction strings using Python f-string syntax like {output_key}. For example, if the first agent sets output_key="summary", the next agent can access it via instruction="Process this summary: {summary}".

What happens if sub-agents in ParallelAgent use the same output_key?

This causes a collision because both agents attempt to write to the same key in the shared session state. Always assign unique output_key values to each sub-agent within a ParallelAgent. The merger agent that follows the parallel execution can then reference both keys to combine the results.

How do I exit a LoopAgent early before max_iterations is reached?

Have a sub-agent return an Event with actions=EventActions(escalate=True). When the LoopAgent processes this event, it terminates the loop immediately. This is commonly used when an evaluator agent determines that a draft passes validation criteria, signaling that refinement is complete.

Can I nest workflow agents inside each other?

Yes. You can nest ParallelAgent or LoopAgent inside SequentialAgent sub-agents lists, and vice versa. The ADK runtime handles the composition recursively. For example, you can place a ParallelAgent inside a SequentialAgent to fetch data concurrently before processing sequentially, or wrap a refinement loop inside a larger pipeline.

{
  "mcpServers": {
    "instagit": {
      "command": "npx",
      "args": ["-y", "instagit@latest"]
    }
  }
}