Adk Workflow Graphs

ADK 2.0 Workflow Graphs

Version 1.0 | Requires google-adk>=2.0.0

ADK 2.0 introduced google.adk.workflow.Workflow — a graph-based orchestration primitive that subsumes SequentialAgent, LoopAgent, and ParallelAgent into one model. Nodes are units of work (functions, LlmAgents, or other BaseNodes); edges declare how output flows and how control routes.

This skill is the API reference. For when to choose a Workflow vs. a single LlmAgent vs. one of the shortcut classes, see adk-agent-patterns.

Minimal example

from google.adk.workflow import Workflow

def greet(node_input: str) -> str:
    return f"Hello! You said: {node_input}"

root_agent = Workflow(
    name="my_workflow",
    edges=[
        ('START', greet),
    ],
)

Run with adk run my_agent/ (CLI) or adk web my_agent/ (web UI).

The 'START' sentinel is the workflow's entry point. ('START', greet) means "when the workflow begins, pass the user input to greet."

References

Load a reference with load_skill_resource("adk-workflow-graphs", "<resource>.md").

Core mental model

A Workflow is itself a BaseNode. Its body is a list of edges, where each edge is a tuple (source, target) and either side can be:

• a single node (function, LlmAgent, FunctionNode, or nested Workflow)
• a tuple of nodes — fan-out (in target) or fan-in (in source)
• a dict mapping route strings to nodes — conditional routing (in target)
• the sentinel 'START' — the entry point (in source only)

edges = [
    ('START', classifier),                       # entry
    (classifier, {                               # routed branches
        "sql":   sql_writer,
        "chart": chart_maker,
        "__DEFAULT__": fallback,
    }),
    ((sql_writer, chart_maker, fallback), join), # fan-in
    (join, format_answer),                       # linear step
]

Output flows along edges automatically: a node's return value becomes the next node's node_input. State that should be visible across the whole graph goes through ctx.state (read in nodes, written by setting ctx.state[key] = val).

Cheat sheet

from google.adk.workflow import Workflow, node, FunctionNode, JoinNode
from google.adk import Event

# 1. Function node (decorator)
@node
def transform(node_input: str) -> str:
    return node_input.upper()

# 2. Function node (explicit, for naming or reuse)
clean = FunctionNode(lambda x: x.strip(), name="clean")

# 3. Routing — return Event(output=..., route=...)
@node
def classify(node_input: str):
    return Event(output=node_input, route="error" if "fail" in node_input else "ok")

# 4. Dict-syntax routing in edges
edges = [
    ('START', classify),
    (classify, {"ok": transform, "error": handle_error}),
]

# 5. Fan-out (tuple on target side)
edges = [('START', (analyze, translate, summarize))]

# 6. Fan-in (tuple on source side, often with JoinNode)
edges = [
    ((analyze, translate, summarize), JoinNode(name="join")),
    (JoinNode, final_step),
]

# 7. Cycle (revision loop) — must include at least one routed exit
edges = [
    ('START', drafter),
    (drafter, reviewer),
    (reviewer, {"revise": drafter, "approve": send}),
]

# 8. Dynamic child node from inside a parent
@node(rerun_on_resume=True)
async def parent(ctx, node_input: str):
    result = await ctx.run_node(child_agent, node_input=node_input)
    return result

Migration from 1.x shortcut classes

SequentialAgent, LoopAgent, and ParallelAgent still work in 2.0 — they are now thin wrappers that compile to the same graph machinery. Migrate to Workflow when:

• You need to branch or route based on a step's result.
• You need to fan out and aggregate (ParallelAgent alone has no join).
• You're nesting two or more of the shortcut classes.
• You want a plain Python step (FunctionNode) inline with LLM steps.

Stay on the shortcut classes for trivially linear (3 steps, no routing) or trivially parallel (no aggregation) cases — they're one line and equivalent.

Critical rules

1. rerun_on_resume=True on any node that calls ctx.run_node(). Without it, an interruption (HITL pause, error retry) won't replay the parent and the dynamic child's result won't be wired back in.
2. Cycles must have a routed edge. (reviewer, drafter) alone is rejected at graph validation — there'd be no way out. Use a dict so at least one route exits the cycle.
3. Use deterministic names for dynamic children — pass name= to ctx.run_node() so replays match the original execution.
4. Don't wrap with asyncio.create_task() inside a node — always await directly. The scheduler handles concurrency; manual tasks break replay.
5. Don't set ctx.event_author manually — the runner manages authorship.

Common mistakes