adk-python-workflows

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](https://adk.dev/workflows/index.md) · [Graph routes](https://adk.dev/graphs/routes/index.md) · [Collaboration](https://adk.dev/workflows/collaboration/index.md) · [Data handling](https://adk.dev/graphs/data-handling/index.md) · [Dynamic workflows](https://adk.dev/graphs/dynamic/index.md) · [Human-in-the-loop](https://adk.dev/graphs/human-input/index.md)

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:

| Python Object | Wrapped As | Default `rerun_on_resume` | |--------------|-----------|------------------------| | Function/callable | `FunctionNode` | `False` | | `LlmAgent` | Internal `_LlmAgentWrapper` | `True` | | Other `BaseAgent` | Internal `AgentNode` | `False` | | `BaseTool` | Internal `_ToolNode` | `False` | | `BaseNode` subclass | Used as-is | Per subclass |

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

| Parameter | Source | |-----------|--------| | `ctx` | Workflow `Context` object | | `node_input` | Output from predecessor node | | Any other name | `ctx.state[param_name]` |

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

| Predecessor | `node_input` Type | |-------------|-------------------| | Function returning `str`/`dict` | `str`/`dict` | | Function returning `Event(output=X)` | type of `X` | | `LlmAgent` (no `output_schema`) | `types.Content` | | `LlmAgent` (with `output_schema`) | `dict` | | `JoinNode` | `dict[str, Any]` (keyed by predecessor names) | | `ParallelWorker` | `list` | | `START` (no `input_schema`) | `types.Content` | | `START` (with `input_schema`) | parsed schema type |

@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),
    (proc