adk-workflow-graphs

name	adk-workflow-graphs
description	Build graph-based agents with ADK 2.0 `Workflow` — declare nodes and edges, route conditionally, fan-out/fan-in in parallel, run dynamic nodes at runtime, and add human-in-the-loop revision cycles. Load this skill whenever the agent needs anything beyond a strictly linear or trivially parallel pipeline.

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

Resource	Description	Load when
function-nodes	@node decorator and FunctionNode; param resolution from ctx.state; async/generator nodes	A graph step is plain Python — transform, validate, branch logic
llm-agent-nodes	_LlmAgentWrapper behavior; output_schema for parsed dicts; tools-in-workflow; emitting routes from LLM output	A graph step is an LlmAgent and you need to compose its output cleanly
routing-and-conditions	Event(route=...), dict-syntax edges, __DEFAULT__ fallback, cyclical edges with validation rules	Branching on a node's result, revision loops, or a state-machine-shaped flow
parallel-and-fanout	Tuple syntax for fan-out/fan-in, JoinNode, diamond patterns, parallel-state-write rules	Running nodes concurrently — especially when you also need to aggregate results
dynamic-nodes	ctx.run_node() API, rerun_on_resume=True, deterministic naming for replay, mixing static + dynamic	The set of next steps depends on runtime data (data-driven fan-out, recursion)

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

Mistake	Fix
Forgetting the 'START' sentinel	Every workflow needs ('START', first_node)
Returning raw types.Content from an LLM node	In 2.0 the wrapper auto-extracts str; just return text or set output_schema= for dict
Unconditional cycle	Add a routed exit: (reviewer, {"revise": drafter, "done": next})
Manual asyncio.create_task() inside a node	Use await ctx.run_node(...) — the scheduler handles parallelism
Two parallel branches writing to the same ctx.state key	Give each branch a unique key, or fan into a JoinNode that merges them
Using ctx.state for per-step output passing	Output flows along edges automatically as node_input; reserve ctx.state for cross-cutting values