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

> Learn to build multi-agent systems with agents-cli and ADK's SequentialAgent, ParallelAgent, and LoopAgent. Discover deterministic workflows and seamless state passing for powerful agent compositions.

- Repository: [Google/agents-cli](https://github.com/google/agents-cli)
- Tags: how-to-guide
- Published: 2026-07-02

---

**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`](https://github.com/google/agents-cli/blob/main/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`](https://github.com/google/agents-cli/blob/main/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.

```python

# 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.

```python

# 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.

```python

# 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`](https://github.com/google/agents-cli/blob/main/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

```bash
adk scaffold my_project
cd my_project

```

This creates the `src/my_project/` directory structure with boilerplate [`agent.py`](https://github.com/google/agents-cli/blob/main/agent.py) and [`app.py`](https://github.com/google/agents-cli/blob/main/app.py) files.

### Define the Application Entry Point

Replace the generated [`src/my_project/agent.py`](https://github.com/google/agents-cli/blob/main/src/my_project/agent.py) with one of the composition patterns shown above. Then ensure [`src/my_project/app.py`](https://github.com/google/agents-cli/blob/main/src/my_project/app.py) wires the root agent:

```python

# 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

```bash
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`](https://github.com/google/agents-cli/blob/main/src/google/agents/cli/eval/_synthesize_runner.py) and [`src/google/agents/cli/eval/_inference_runner.py`](https://github.com/google/agents-cli/blob/main/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:

```python
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:

```python

# 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,
        ),
    ],
)

```

```python

# 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:

```bash
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`](https://github.com/google/agents-cli/blob/main/skills/google-agents-cli-adk-code/references/adk-python.md) and [`skills/google-agents-cli-adk-code/SKILL.md`](https://github.com/google/agents-cli/blob/main/skills/google-agents-cli-adk-code/SKILL.md).
- Use `adk scaffold` to create projects, define your composition in [`src/my_project/agent.py`](https://github.com/google/agents-cli/blob/main/src/my_project/agent.py), wire the root agent in [`app.py`](https://github.com/google/agents-cli/blob/main/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.