/google-agents-cli-adk-code

ADK Workflow API Cheatsheet

Requires google-adk >= 2.0.0. This page documents the Python graph API; ADK Go has its own — see references/adk-go-workflows.md. Requires Python >= 3.11. A Workflow can also run as a Live (voice) agent with no structural change — give the stage agents a Live model; see references/adk-python-live.md.

Official docs: Workflows overview · Graph routes · Collaboration · Data handling · Dynamic workflows · Human-in-the-loop

1. Core Concepts

A Workflow is a graph-based agent: nodes do work, edges define flow, START is the entry point.

from google.adk.workflow import Workflow

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

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

Three building blocks: Nodes (functions, LLM agents, tools), Edges (connections with optional route conditions), START (built-in entry receiving user input).

Workflow Constructor

root_agent = Workflow(
    name="my_workflow",
    edges=[...],                # Edge definitions (or use graph= instead)
    description="",             # Agent description
    input_schema=None,          # Pydantic model for input validation
    output_schema=None,         # Pydantic model for the workflow's output
    state_schema=None,          # Pydantic model for state validation
    rerun_on_resume=True,       # Rerun workflow on resume (default: True)
    max_concurrency=None,       # Limit parallel node execution (None = no limit)
    retry_config=None,          # Default RetryConfig applied to nodes
    timeout=None,               # Whole-workflow timeout in seconds
    wait_for_output=False,      # Wait for dynamically scheduled child output
)

2. Node Types

Any "NodeLike" is accepted in edges and auto-wrapped:

Auto-wrapping is the recommended approach. Place functions, agents, and tools directly in edges — the framework wraps them automatically. You do not need to import or use internal wrapper classes directly.

3. Function Nodes

Most common node type. Parameter resolution:

from google.adk.agents.context import Context

def process(ctx: Context, node_input: Any, user_name: str) -> str:
  # node_input = predecessor output; user_name = ctx.state['user_name']
  # START outputs types.Content (not str) unless input_schema is set
  return f"{user_name}: {node_input}"

Return Types

• Value -> wrapped in Event(output=value), triggers downstream
• None -> no event emitted, no downstream trigger
• Event -> used directly (for routing or state updates)
• Generator -> yield multiple events; only the last with output triggers downstream

from google.adk.events.event import Event

def classify(node_input: str):
  if "urgent" in node_input:
    return Event(output=node_input, route="urgent")
  return Event(output=node_input, route="normal", state={"processed": True})

Auto Type Conversion

FunctionNode auto-converts dict inputs to Pydantic models based on type hints. Works for list[Model] and dict[str, Model] too.

node_input Type by Predecessor

@node Decorator & Explicit FunctionNode

from google.adk.workflow import node, FunctionNode, RetryConfig

@node
def my_func(node_input: str) -> str:
  return node_input

@node(name="custom", rerun_on_resume=True)
async def my_async(node_input: str) -> str:
  return node_input

# Explicit FunctionNode for full control (func is keyword-only)
fn = FunctionNode(
    func=my_func,
    retry_config=RetryConfig(max_attempts=3),
    timeout=30.0,                          # Seconds before timeout
    parameter_binding='state',             # 'state' (default) or 'node_input'
    auth_config=None,                      # Requires rerun_on_resume=True
    state_schema=None,                     # Pydantic model for state validation
)

4. Edge Patterns

# Sequential chain
edges = [('START', a), (a, b), (b, c)]

# Conditional routing (node returns Event with route=)
edges = [
    ('START', classifier),
    (classifier, success_handler, "success"),
    (classifier, error_handler, "error"),
    (classifier, fallback_handler, '__DEFAULT__'),  # Fallback route
]

# Fan-out (parallel branches)
edges = [('START', (branch_a, branch_b, branch_c))]

# Fan-in with JoinNode
from google.adk.workflow import JoinNode
join = JoinNode(name="merge")
edges = [((branch_a, branch_b), join), (join, final)]
# JoinNode output: {"branch_a": output_a, "branch_b": output_b}

# Looping (must have at least one routed edge — unconditional cycles rejected)
edges = [
    ('START', process),
    (process, check),
    (check, process, "continue"),
    (check, finish, "exit"),
]

Route values: str, bool, int. Multi-route fan-out: return Event(output=x, route=["a", "b"]). Edge matching multiple routes: (node, target, ["route_x", "route_y"]).

5. LLM Agent Nodes

Use google.adk.agents.LlmAgent in workflow edges — auto-wrapped internally, emits Event(output=...) for downstream data passing.

from google.adk.agents import LlmAgent
from pydantic import BaseModel

class DraftOutput(BaseModel):
  title: str
  content: str

writer = LlmAgent(
    name="writer",
    model="gemini-3.8-flash",
    instruction="Write a draft based on the user's request.",
    output_schema=DraftOutput,  # Always set for structured output
    output_key="draft",         # Also store in state['draft']
)

agent = Workflow(
    name="pipeline",
    edges=[('START', writer), (writer, process_draft)],
)

Always use output_schema (Pydantic model) on LLM agents in workflows. Without it, output is types.Content which may cause type errors in downstream function nodes or serialization failures with JoinNode/database sessions.

6. Parallel Processing

ParallelWorker — process list items concurrently

from google.adk.workflow import node

@node(parallel_worker=True)
def process_item(node_input: int) -> int:
  return node_input * 2

# Input: [1, 2, 3] -> Output: [2, 4, 6]
agent = Workflow(
    name="parallel",
    edges=[('START', split_input), (split_input, process_item), (process_item, collect)],
)

Workers named {parent_name}@{index}. Input must be a list. Output is a list in same order.

Fan-Out / Fan-In — diamond pattern

from google.adk.workflow import JoinNode
join = JoinNode(name="merge")
edges = [
    ('START', splitter),
    (splitter, (branch_a, branch_b)),
    ((branch_a, branch_b), join),
    (join, combiner),  # combiner receives {"branch_a": ..., "branch_b": ...}
]

7. Human-in-the-Loop (HITL)

HITL is a general ADK feature (app-level ResumabilityConfig, the request_input tool, resume_inputs) — see adk-python.md "Human-in-the-Loop". Inside a workflow it works per node: a node yields RequestInput and reads replies from ctx.resume_inputs (keyed by interrupt_id).

from google.adk.events.request_input import RequestInput

async def multi_step(ctx: Context, node_input: str):
  if not ctx.resume_inputs:
    yield RequestInput(interrupt_id="ask_name", message="Name?")
    return
  if "ask_email" not in ctx.resume_inputs:
    yield RequestInput(interrupt_id="ask_email", message="Email?")
    return
  yield Event(output={"name": ctx.resume_inputs["ask_name"],
                      "email": ctx.resume_inputs["ask_email"]})

Node resume behavior: rerun_on_resume=False (default FunctionNode) → the user's response becomes the node output; rerun_on_resume=True (default LlmAgent) → the node reruns with ctx.resume_inputs populated. In loops: use a unique interrupt_id per iteration (e.g. f'review_{count}') to avoid infinite restarts.

8. State & Events

Context Properties

from google.adk.agents.context import Context

def my_node(ctx: Context, node_input: str) -> str:
  ctx.state.get("key", "default")   # Read state
  ctx.session.id                     # Session ID
  ctx.node_path                      # "Workflow/node_name"
  ctx.node                           # Current node
  ctx.run_id                         # Current execution ID
  ctx.attempt_count                  # 1 on first attempt (1-based)
  ctx.resume_inputs                  # HITL resume data (dict keyed by interrupt_id)
  ctx.interrupt_ids                  # Active interrupt IDs
  ctx.output                         # Node's result value (settable)
  ctx.route                          # Routing value (settable)
  return "result"

Dynamic Node Scheduling

async def orchestrator(ctx: Context, node_input: list) -> list:
  results = []
  for i, item in enumerate(node_input):
    result = await ctx.run_node(process_item, node_input=item)
    results.append(result)
  return results

ctx.run_node() requires rerun_on_resume=True on the calling node. Use use_as_output=True to delegate the node's output to the dynamic child.

Event Fields

State: Prefer Event over ctx.state

# Preferred — persisted in event history, replayable
def save(node_input: str):
  return Event(output=node_input, state={"key": node_input})

# Avoid — side effect, may be lost on replay
def save(ctx: Context, node_input: str) -> str:
  ctx.state["key"] = node_input
  return node_input

Data Serialization Rules

• Event.output must be JSON-serializable. BaseModel returns auto-converted via model_dump().
• output_key stores dicts (not BaseModel instances) — validate_schema() -> model_dump().
• ctx.state.get(key) returns a dict. Use MyModel(**data) to reconstruct typed access.

9. Retry Configuration

from google.adk.workflow import FunctionNode, RetryConfig

node = FunctionNode(
    func=flaky_call,           # func is keyword-only
    retry_config=RetryConfig(
        max_attempts=5,        # Default: None (treated as 5); 1 = no retry
        initial_delay=1.0,     # Seconds before first retry
        max_delay=60.0,        # Max seconds between retries
        backoff_factor=2.0,    # Delay multiplier per attempt
        jitter=1.0,            # Randomness factor (0.0 = none)
        exceptions=None,       # Exception types to retry (None = all)
    ),
)

Delay formula: min(initial_delay * backoff_factor^attempt, max_delay) * (1 + random(0, jitter))

10. Testing

Note: The testing utilities below (testing_utils, InMemoryRunner) are internal to the ADK repository. They are not part of the public google-adk package. For your own tests, use App + InMemoryRunner from google.adk.runners or write a custom test harness.

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

@pytest.mark.asyncio
async def test_workflow():
  def step(node_input: str) -> str:
    return "done"

  agent = Workflow(name="test", edges=[('START', step)])
  app = App(name="test_app", root_agent=agent)
  runner = InMemoryRunner(app=app)

  session = await runner.session_service.create_session(
      app_name="test_app", user_id="test_user"
  )
  async for event in runner.run_async(
      user_id="test_user",
      session_id=session.id,
      new_message=types.Content(role="user", parts=[types.Part.from_text(text="hello")]),
  ):
    if event.output is not None:
      assert event.output == "done"

11. Import Paths

Workflow Core

Workflow Nodes (auto-wrapped)

Nodes are auto-wrapped when placed in edges. You do not need to import wrapper classes.

Events & Context

LLM Agent

App & Resumability

12. Best Practices

Use Pydantic Models, Not Raw Dicts

Always define BaseModel classes for node I/O, LLM output_schema, and structured data:

# Wrong: raw dicts
def lookup(node_input: dict[str, Any]) -> dict[str, Any]:
  return {"cost": 500}

# Correct: typed schemas
class FlightInfo(BaseModel):
  cost: int
  details: str

def lookup(node_input: Itinerary) -> FlightInfo:
  return FlightInfo(cost=500, details="Economy")

Emit Content Events for Web UI

event.output is internal — only event.content renders in the ADK web UI:

from google.genai import types

def final_output(node_input: str):
  yield Event(content=types.Content(role='model', parts=[types.Part.from_text(text=node_input)]))
  yield Event(output=node_input)

LLM agents emit content events automatically. Add them explicitly for function nodes with user-facing results.

Agent Directory Convention

my_workflow/
  __init__.py    # from . import agent
  agent.py       # root_agent = Workflow(...)

Advanced Patterns

• Nested workflows: A Workflow can be used as a node in another workflow
• Dynamic node scheduling: Use await ctx.run_node(func, node_input=item) at runtime (requires rerun_on_resume=True)
• Custom Node subclass: Subclass Node, implement run_node_impl(*, ctx, node_input) -> AsyncGenerator. Supports parallel_worker=True flag.
• Custom BaseNode: Subclass BaseNode, implement _run_impl(*, ctx, node_input) -> AsyncGenerator

Graph Validation Rules

1. START must exist and have no incoming edges
2. All non-START nodes must be reachable
3. No duplicate node names or edges
4. At most one __DEFAULT__ route per node
5. No unconditional cycles (cycles need at least one routed edge)

Further Reading

• Workflows overview
• Graph routes & conditional edges
• Agent collaboration & task mode
• Data handling & state
• Dynamic workflows
• Human-in-the-loop
• App class (workflow container)