Adk Agent Builder

ADK Agent Builder

This file serves as a directory of specialized reference guides for developing agents with ADK. To avoid context pollution, read only the relevant reference file based on your current task.

Core Concepts Directory

Refer to these files for foundational knowledge:

• Getting Started & Basic Agents: getting-started.md
• Environment setup, API key configuration, and minimal agent definitions.
• Tool Catalog: tool-catalog.md
• How to bind function tools, MCP tools, OpenAPI specs, and Google API tools.
• Agent Modes (Task / Single-Turn): task-mode.md
• Multi-turn structured delegation and autonomous single-turn execution patterns.

Workflow & Graph Orchestration

Refer to these files when building complex graphs:

• Function Nodes: function-nodes.md
• How to use functions as nodes, type resolution, and generators.
• Routing & Conditions: routing-and-conditions.md
• Edge patterns, dict-based routing, self-loops, and conditional execution.
• LLM Agent Nodes: llm-agent-nodes.md
• How to use LLM agents as workflow nodes, task wrappers, and handling output schemas.

Advanced Orchestration Patterns

• Parallel Processing & Fan-Out: parallel-and-fanout.md
• ParallelWorker for list splitting and concurrent processing, fan-out/join patterns.
• Human-in-the-Loop: human-in-the-loop.md
• Pausing execution for user input, resumable workflows, and AuthConfig on nodes.
• Dynamic Nodes: dynamic-nodes.md
• Scheduling nodes at runtime dynamically via ctx.run_node().

Infrastructure & Utilities

• State & Events: state-and-events.md
• Using context API, sharing global state, and yield event structures.
• Multi-Agent Systems: multi-agent.md
• Hierarchical execution (e.g., SequentialAgent, LoopAgent, ParallelAgent).
• Testing Strategies: testing.md
• Automated queries with adk run, unit tests, and integration testing with sample agents.

Standards & Guidelines

• Best Practices: best-practices.md
• Critical rules (Pydantic schemas, content events, state-based data flow).

Advanced Workflow Patterns Reference

Nested workflows, dynamic nodes, retry configuration, custom node types, and graph construction.

📋 Agent Verification Checklist (Advanced Patterns)

Use this checklist when implementing complex workflows:

• [ ] Validation: Does your graph follow all 7 validation rules? (e.g., no unconditional cycles)
• [ ] Custom Nodes: If creating a custom node, did you override get_name() and run()?
• [ ] Dynamic Execution: If using run_node, did you follow the rules in the dedicated dynamic-nodes reference?
• [ ] Waiting State: Did you use wait_for_output=True if the node should stay in WAITING state until output is yielded?

💡 Quick Reference

• Retry: RetryConfig(max_attempts=5, initial_delay=1.0)
• Custom Node Fields: rerun_on_resume, wait_for_output, retry_config, timeout

Nested Workflows

A Workflow is both an agent and a node. Use one workflow inside another:

from google.adk.workflow import Workflow

# Inner workflow
inner = Workflow(
    name="inner_pipeline",
    edges=[
        ('START', step_a),
        (step_a, step_b),
    ],
)

# Outer workflow using inner as a node
outer = Workflow(
    name="outer_pipeline",
    edges=[
        ('START', pre_process),
        (pre_process, inner),      # Nested workflow
        (inner, post_process),
    ],
)

The inner workflow receives the predecessor's output as its START input and its terminal output flows to the next node in the outer workflow.

Dynamic Node Scheduling

Schedule nodes at runtime using ctx.run_node().

See the dedicated Dynamic Node Scheduling Reference for detailed rules, examples, and best practices.

Retry Configuration

Configure automatic retry for nodes that may fail:

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

retry = RetryConfig(
    max_attempts=5,         # Max attempts (default: 5). 0 or 1 = no retry
    initial_delay=1.0,      # Seconds before first retry (default: 1.0)
    max_delay=60.0,         # Max seconds between retries (default: 60.0)
    backoff_factor=2.0,     # Delay multiplier per attempt (default: 2.0)
    jitter=1.0,             # Randomness factor (default: 1.0, 0.0 = none)
    exceptions=None,        # Exception types to retry (None = all)
)

node = FunctionNode(
    flaky_api_call,
    name="api_call",
    retry_config=retry,
)

Retry delay formula

delay = initial_delay * (backoff_factor ^ attempt)
delay = min(delay, max_delay)
delay = delay * (1 + random(0, jitter))

Accessing the attempt count

def my_node(ctx: Context, node_input: str) -> str:
  # attempt_count is 1 on the first try, ≥2 on retries
  if ctx.attempt_count > 1:
    print(f"Retry attempt {ctx.attempt_count}")
  return "result"

Custom Node Types

Subclass BaseNode for custom behavior:

from google.adk.workflow import BaseNode
from google.adk.events.event import Event
from google.adk.agents.context import Context
from pydantic import ConfigDict, Field
from typing import Any, AsyncGenerator
from typing_extensions import override

class BatchProcessorNode(BaseNode):
  """Processes items in batches."""
  model_config = ConfigDict(arbitrary_types_allowed=True)

  name: str = Field(default="batch_processor")
  batch_size: int = Field(default=10)

  def __init__(self, *, name: str = "batch_processor", batch_size: int = 10):
    super().__init__()
    object.__setattr__(self, 'name', name)
    object.__setattr__(self, 'batch_size', batch_size)

  @override
  def get_name(self) -> str:
    return self.name

  @override
  async def run(
      self,
      *,
      ctx: Context,
      node_input: Any,
  ) -> AsyncGenerator[Any, None]:
    items = node_input if isinstance(node_input, list) else [node_input]
    results = []
    for i in range(0, len(items), self.batch_size):
      batch = items[i:i + self.batch_size]
      batch_result = await process_batch(batch)
      results.extend(batch_result)
    yield Event(output=results)

BaseNode Fields

wait_for_output

When wait_for_output=True, a node that finishes without yielding an Event with output moves to WAITING state instead of COMPLETED. Downstream nodes are not triggered. The node can then be re-triggered by upstream predecessors.

This is how JoinNode works internally — it runs once per predecessor, storing partial inputs, and only yields output (triggering downstream) when all predecessors have completed. LlmAgentWrapper in task mode also sets wait_for_output=True automatically.

from google.adk.workflow import BaseNode

class CollectorNode(BaseNode):
  wait_for_output: bool = True  # Stay in WAITING until output is yielded

  async def run(self, *, ctx, node_input):
    # Store partial input, don't yield output yet
    collected = ctx.state.get("collected", [])
    collected.append(node_input)
    yield Event(state={"collected": collected})

    # Only yield output when we have enough
    if len(collected) >= 3:
      yield Event(output=collected)
      # Now node transitions to COMPLETED and triggers downstream

Nodes with wait_for_output=True default:

• JoinNode: True (waits for all predecessors)
• LlmAgentWrapper (task mode): True (set in model_post_init)
• All other nodes: False

Required Methods

ToolNode

Wrap an ADK tool as a workflow node:

from google.adk.workflow._tool_node import _ToolNode as ToolNode
from google.adk.tools.function_tool import FunctionTool

def search(query: str) -> str:
  """Search for information."""
  return f"Results for: {query}"

tool = FunctionTool(search)
tool_node = ToolNode(tool, name="search_node")

agent = Workflow(
    name="with_tool",
    edges=[
        ('START', prepare_query),
        (prepare_query, tool_node),  # Input must be dict (tool args) or None
        (tool_node, process_results),
    ],
)

Important: ToolNode input must be a dictionary of tool arguments or None.

AgentNode

Wrap any BaseAgent (not just LlmAgent) as a workflow node:

from google.adk.workflow._agent_node import AgentNode
from google.adk.agents.loop_agent import LoopAgent

loop = LoopAgent(
    name="refine_loop",
    sub_agents=[writer, reviewer],
    max_iterations=3,
)

loop_node = AgentNode(agent=loop, name="refinement")

agent = Workflow(
    name="with_loop",
    edges=[
        ('START', loop_node),
        (loop_node, final_step),
    ],
)

Graph Validation Rules

The workflow graph is validated on construction. These rules are enforced:

1. START node must exist 2. START node must not have incoming edges 3. All non-START nodes must be reachable (appear as to_node in some edge) 4. No duplicate node names 5. No duplicate edges 6. At most one __DEFAULT__ route per node 7. No unconditional cycles (cycles must have at least one routed edge)

Edge Construction Patterns

from google.adk.workflow import Edge
from google.adk.workflow._workflow_graph import WorkflowGraph

# Tuple syntax (most common)
edges = [
    ('START', node_a),                    # Simple edge
    (node_a, node_b, "route"),            # Routed edge
    (node_a, (node_b, node_c)),           # Fan-out
    ((node_b, node_c), join_node),        # Fan-in
]

# Sequence shorthand (tuple with 3+ elements creates chain)
edges = [('START', node_a, node_b, node_c)]
# Equivalent to: [('START', node_a), (node_a, node_b), (node_b, node_c)]

# Routing map (dict syntax)
edges = [
    (classifier, {"success": handler_a, "error": handler_b}),
]

# Edge objects (explicit)
edges = [
    Edge(START, node_a),
    Edge(node_a, node_b, route="success"),
]

# Edge.chain helper
edges = Edge.chain('START', node_a, node_b, node_c)
# Returns: [(START, node_a), (node_a, node_b), (node_b, node_c)]

# WorkflowGraph.from_edge_items
graph = WorkflowGraph.from_edge_items([
    ('START', node_a),
    (node_a, node_b),
])
agent = Workflow(name="my_workflow", graph=graph)

Source File Locations

ADK Workflow Best Practices

This document outlines the critical best practices and rules for developing reliable and maintainable workflows with the ADK.

📋 Agent Code Verification Checklist

Use this checklist to verify your code before submitting or finalizing changes:

• [ ] Schemas: Are Pydantic BaseModel classes used for all inputs/outputs? (No raw dicts)
• [ ] UI Output: Do user-visible messages use Event(message=...)? (Not output=)
• [ ] State Data Flow: Is data stored in state and read via {var} or param names?
• [ ] State Updates: Are state updates done via Event(state=...)? (Avoid direct ctx.state mutation)
• [ ] Outputs: Does each node execution yield at most one event.output?
• [ ] Semantics: Are yield and return never mixed in the same function?
• [ ] Instructions: Are {node_input} templates NOT used in agent instructions?
• [ ] HITL: Are interrupt_ids unique per iteration in loops?

Best Practices (MUST FOLLOW)

Use Pydantic Models, Not Raw Dicts

Always define Pydantic `BaseModel` classes for function node inputs, outputs, LLM output_schema, and structured data. Never use dict[str, Any] when the shape is known:

# ❌ WRONG: raw dicts
def lookup_flights(node_input: dict[str, Any]) -> dict[str, Any]:
  return {"flight_cost": 500, "details": "Economy"}

# ✅ CORRECT: typed schemas
class FlightInfo(BaseModel):
  flight_cost: int
  details: str

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

This applies to ALL data flowing through the graph: node inputs, node outputs, JoinNode results, LLM output schemas, and HITL response schemas.

Emit Content Events for Web UI Display

event.output is internal — only event.content renders in the ADK web UI. For user-visible output, use Event(message=...):

def final_output(node_input: str):
  yield Event(message=node_input)  # message= renders in web UI
  yield Event(output=node_input)   # output= passes data to downstream nodes

# State-only event (no output, no message — just side-effect state update)
def store_data(node_input: str):
  yield Event(state={"user_input": node_input})

> [!TIP]
> Function nodes can stream user-visible messages by yielding `Event(message="chunk", partial=True)`.

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

Prefer State-Based Data Flow with LLM Agents

Store data in state via Event(state={...}) or output_key, then read it via instruction templates {var} or function parameter name injection. This is more robust than passing data through node_input, especially for routing workflows where multiple branches need the same data.

# ✅ State-based: store early, read anywhere via {var} or param name
def process_input(node_input: str):
  yield Event(state={"topic": node_input})

writer = Agent(name="writer", instruction='Write about "{topic}".', output_key="draft")
def send(draft: str):  # draft resolved from ctx.state["draft"]
  yield Event(message=draft)

# ❌ Fragile: threading data through node_input breaks at routing/loops

Set State via Event, Not ctx.state

Prefer `Event(state=...)` over `ctx.state[key] = ...` for writing state. Event-based state is persisted in event history and replayable during non-resumable HITL. Direct ctx.state mutations are side effects that may be lost on replay.

# ✅ Preferred
def save(node_input: str):
  return Event(output=node_input, state={"user_request": node_input})

# ❌ Avoid
def save(ctx: Context, node_input: str) -> str:
  ctx.state["user_request"] = node_input
  return node_input

One Output Event Per Node

Each node execution can yield many events, but at most one should have `event.output`. This applies to function nodes, LLM agents (including task and single_turn mode), and nested workflows. Multiple output events get silently merged into a list, which changes the downstream node_input type and usually causes errors. Similarly, at most one event can have route — multiple routed events raise ValueError.

# ✅ Correct: one output event, other events for messages/state
def my_node(node_input: str):
  yield Event(message="Processing...")      # display only
  yield Event(state={"status": "done"})     # state update only
  yield Event(output="final result")        # the single output

# ❌ Wrong: multiple output events
def my_node(node_input: str):
  yield Event(output="first")   # these get merged into ["first", "second"]
  yield Event(output="second")  # downstream expects str, gets list → TypeError

Don't Mix yield and return Event

A function is either a generator (uses yield) or a regular function (uses return). Never mix them — in Python, a function with yield becomes a generator and any return value is silently ignored:

# ✅ Generator: use yield for all events
def my_node(node_input: str):
  yield Event(state={"key": "value"})
  yield Event(output="result")

# ✅ Regular function: use return for a single value/event
def my_node(node_input: str):
  return Event(output="result", state={"key": "value"})

# ✅ Regular function: return plain value (auto-wrapped in Event)
def my_node(node_input: str) -> str:
  return "result"

# ❌ Wrong: mixing yield and return — the return is silently ignored
def my_node(node_input: str):
  yield Event(state={"key": "value"})
  return Event(output="result")  # IGNORED — Python generator semantics

Use generators (yield) when you need multiple events (state + output + message). Use regular functions (return) for simple single-value output.

Never Put node_input in LLM Agent Instructions

{var} templates in instruction resolve only from ctx.state. node_input is NOT available as a template variable — it is automatically sent as the user message to the LLM. Do not try to reference it in the instruction:

# ❌ Wrong: {node_input} is not in state, raises KeyError
agent = Agent(
    name="summarizer",
    instruction="Summarize this: {node_input}",
)

# ✅ Correct: node_input already becomes the user message, just instruct
agent = Agent(
    name="summarizer",
    instruction="Summarize the following text in one sentence.",
)

# ✅ Correct: use state for data that needs to be in the instruction
agent = Agent(
    name="writer",
    instruction='Write about "{topic}". Previous feedback: {feedback?}',
    output_key="draft",
)

Workflow Cannot Be a Sub-Agent of LlmAgent

Workflow, SequentialAgent, LoopAgent, and ParallelAgent cannot be added as sub_agents of an LlmAgent. Agent transfer to workflow agents is not supported.

Workflow Data Rules

• `Event.output` must be JSON-serializable. FunctionNode auto-converts BaseModel returns via model_dump(). Never store types.Content or other non-serializable objects in Event.output.
• `output_key` stores dicts, not BaseModel instances. LLM agents with output_schema run validate_schema() → model_dump(), so ctx.state[output_key] is a plain dict.
• `ctx.state.get(key)` returns a dict. Use dict access (data["field"]) or reconstruct (MyModel(**data)) for typed access.

Human-in-the-Loop (HITL) Rules

Unique interrupt_id in Loops

When a node requests input (yields RequestInput) inside a loop (e.g., a review-revise loop), you MUST use a unique `interrupt_id` per iteration (e.g., review_{count}).

If you reuse the same interrupt_id, the event-based state reconstruction will confuse responses from earlier iterations with the current one, leading to infinite restart loops!

# ✅ Correct: unique ID per iteration
review_count = ctx.state.get('review_count', 0)
interrupt_id = f'review_{review_count}'
yield RequestInput(interrupt_id=interrupt_id, message="Approve?")

Callbacks and Plugins

📋 Agent Verification Checklist (Callbacks)

Use this checklist when implementing callbacks or plugins:

• [ ] Override Behavior: Remember that returning a non-None value in a callback overrides the default behavior (e.g., skips model call or tool execution). Is that intentional?
• [ ] Context Type: Remember that CallbackContext is an alias for Context.

💡 Quick Reference (Callback Returns)

• Continue Normal Flow: Return None.
• Override Model: Return LlmResponse in before_model.
• Override Tool: Return dict in before_tool.

Agent Callbacks

root_agent = Agent(
    before_agent_callback=my_before_cb,      # Before agent runs
    after_agent_callback=my_after_cb,         # After agent runs
    before_model_callback=my_before_model,    # Before LLM call
    after_model_callback=my_after_model,      # After LLM call
    before_tool_callback=my_before_tool,      # Before tool call
    after_tool_callback=my_after_tool,        # After tool call
    on_model_error_callback=my_error_cb,      # On LLM error
    on_tool_error_callback=my_tool_error_cb,  # On tool error
    ...
)

Note: CallbackContext is a backward-compatible alias for Context. Both work identically.

Callback Signatures

# before_agent / after_agent
def callback(callback_context: CallbackContext):
  return None  # Continue normal flow
  # OR return ModelContent to override

# before_model
def callback(callback_context, llm_request: LlmRequest):
  return None  # Continue to LLM
  # OR return LlmResponse to skip LLM

# after_model
def callback(callback_context, llm_response):
  return None  # Use actual response
  # OR return LlmResponse to override

# before_tool
def callback(tool, args, tool_context):
  return None  # Call tool normally
  # OR return dict to skip tool

# after_tool
def callback(tool, args, tool_context, tool_response):
  return None  # Use actual response
  # OR return dict to override

Multiple callbacks: Pass a list. They execute in order until one returns non-None.

Plugins (App-Level Callbacks)

from google.adk.plugins.base_plugin import BasePlugin

class MyPlugin(BasePlugin):
  def __init__(self):
    super().__init__(name='my_plugin')

  async def before_agent_callback(self, *, agent, callback_context):
    pass

  async def before_model_callback(self, *, callback_context, llm_request):
    pass

Built-in Plugins

Usage with App:

from google.adk.apps import App
from google.adk.plugins.context_filter_plugin import ContextFilterPlugin

app = App(
    name='my_app',
    root_agent=root_agent,
    plugins=[ContextFilterPlugin(num_invocations_to_keep=3)],
)

Dynamic Node Scheduling Reference

Schedule nodes at runtime using ctx.run_node(). This allows a node within a workflow to trigger the run of another node (or a callable that can be built into a node) and asynchronously wait for its result.

📋 Agent Verification Checklist (Dynamic Nodes)

Use this checklist when scheduling nodes dynamically:

• [ ] Rerun on Resume: Does the parent node calling run_node have rerun_on_resume=True?
• [ ] Run ID: If using an explicit run_id, does it contain non-numeric characters?
• [ ] Param Name: If passing input directly to a raw function via node_input=..., is that function's parameter named node_input?
• [ ] Nesting: If the child node also calls run_node, is it wrapped in FunctionNode(..., rerun_on_resume=True)?

💡 Quick Reference

• Call: await ctx.run_node(node_like, node_input=...)
• Output Delegation: Set use_as_output=True to make child output be the parent's output.

Basic Usage

from google.adk import Agent, Context, Event, Workflow
from google.adk.workflow import node
from pydantic import BaseModel

class Feedback(BaseModel):
  grade: str

generate_headline = Agent(
    name="generate_headline",
    instruction='Write a headline about the topic "{topic}".',
)

evaluate_headline = Agent(
    name="evaluate_headline",
    instruction="Grade whether the headline is tech-related.",
    output_schema=Feedback,
    mode="single_turn",
)

@node(rerun_on_resume=True)
async def orchestrate(ctx: Context, node_input: str) -> str:
  yield Event(state={"topic": node_input})
  while True:
    headline = await ctx.run_node(generate_headline)
    feedback = Feedback.model_validate(
        await ctx.run_node(evaluate_headline, node_input=headline)
    )
    if feedback.grade == "tech-related":
      yield headline
      break

root_agent = Workflow(
    name="root_agent",
    edges=[("START", orchestrate)],
)

Requirements & Rules

• `rerun_on_resume=True`: The parent node calling ctx.run_node() must have rerun_on_resume=True. This is required because dynamically scheduled nodes might be interrupted (e.g., for HITL), and the workflow needs to wake up and re-run the parent node to get the child node's response.
• Unique Instance Names: Each dynamic instance needs a unique name (auto-generated for Agent nodes).
• Node-Like Acceptable: ctx.run_node() accepts any node-like object (function, Agent, BaseNode).
• Explicit `run_id` Constraint: If you provide an explicit run_id, it must contain non-numeric characters (e.g., "run_a" instead of "1") to prevent collision with auto-generated numeric IDs.
• `use_as_output=True`: Suppresses the parent node's own output and uses the child's output as the parent's output. This is achieved via outputFor annotation in events. This can only be called ONCE per parent node execution.
• `use_sub_branch`: (Optional) If set to True, attaches a branch segment (node_name@run_id) to the current execution branch to ensure event isolation for parallel or sub-agent runs.

Best Practices

• Always await ctx.run_node() directly. Wrapping it in asyncio.create_task() means the task runs unsupervised — errors are silently swallowed and the task is not cancelled if the parent node is interrupted.

Imperative Workflow Construction

As an alternative to defining static graph edges, you can use dynamic nodes to construct workflows in an imperative style using standard Python control flow. This approach can sometimes be more intuitive for complex conditional logic or parallel execution.

Replacing Graph Patterns

1. Sequences & Branching

Instead of defining edges with routes, use standard Python if/else:

async def orchestrator(ctx: Context, node_input: str):
  res_a = await ctx.run_node(step_a, node_input=node_input)
  if "success" in res_a:
    return await ctx.run_node(step_b, node_input=res_a)
  else:
    return await ctx.run_node(step_c, node_input=res_a)

Important Pits & Best Practices

• Function Parameter Mapping: When passing a raw function to run_node, ADK defaults to 'state' binding mode. If you want to pass input directly via node_input=... in run_node, the function parameter MUST be named `node_input`!

  def my_worker(node_input: str): # MUST be named 'node_input'
    return f"Done: {node_input}"

• Nested Dynamic Nodes: If a dynamically scheduled node itself calls run_node, it acts as a parent node and MUST have `rerun_on_resume=True`! Since raw functions passed to run_node default to False, you must manually wrap the inner parent function in FunctionNode(..., rerun_on_resume=True)!
• Generator Returns: In nodes that use yield (generators), you cannot use return value to produce the final output (Python syntax error in async generators). You must yield Event(output=...) instead.

Function Nodes Reference

Function nodes are the most common node type. Any Python function becomes a workflow node.

📋 Agent Verification Checklist (Function Nodes)

Use this checklist to verify your Function Node configuration:

• [ ] Input Type: If following an LLM agent without schema, is node_input typed as Any or types.Content? (Not str)
• [ ] UI Output: Do you yield Event(message=...) for results that should appear in the Web UI?
• [ ] Outputs: Does the function yield or return at most one event.output?
• [ ] Union Types: If using Union types for node_input, did you add isinstance checks in the body for actual validation?

💡 Quick Reference (Param Resolution)

• `ctx`: Workflow Context object.
• `node_input`: Output from the predecessor node.
• Any other name: Auto-resolved from ctx.state[param_name].

Imports

from google.adk.workflow import FunctionNode
from google.adk.events.event import Event
from google.adk.agents.context import Context
from google.adk.workflow import node  # @node decorator

Basic Functions

A function returning a value automatically wraps it in an Event:

def process(node_input: str) -> str:
  return f"Processed: {node_input}"

# Async functions work too
async def fetch_data(node_input: str) -> dict:
  result = await some_api_call(node_input)
  return {"data": result}

Function Signatures

FunctionNode inspects the function signature to resolve parameters:

# Receives both context and input
def my_node(ctx: Context, node_input: str) -> str:
  session_id = ctx.session.id
  return f"Session {session_id}: {node_input}"

# Receives only input
def simple(node_input: str) -> str:
  return node_input.upper()

# Reads from state (other params resolved from ctx.state)
def uses_state(node_input: str, user_name: str) -> str:
  # user_name read from ctx.state['user_name']
  return f"{user_name}: {node_input}"

# No parameters at all
def constant() -> str:
  return "hello"

Generator Functions

Yield multiple events from a single node:

# Async generator
async def multi_output(ctx: Context) -> AsyncGenerator[Any, None]:
  yield Event(output="first output")
  yield Event(output="second output")

# Sync generator
def sync_multi(node_input: str):
  yield Event(output="step 1")
  yield Event(output="step 2")

At most one event should have `output`. Multiple output events get silently merged into a list, changing the downstream type. Similarly, at most one event can have route (multiple raise ValueError). Use separate events for messages, state updates, and the single output.

Yielding Raw Values

Yield raw values instead of Event objects. They are wrapped automatically:

async def raw_yield(node_input: str):
  yield "output value"  # Wrapped in Event(output="output value")

Returning None

If a function returns None, no event is emitted and no downstream node is triggered:

def maybe_output(node_input: str) -> str | None:
  if not node_input:
    return None  # No downstream trigger
  return f"Got: {node_input}"

Auto Type Conversion

FunctionNode automatically converts dict inputs to Pydantic models based on type hints:

from pydantic import BaseModel

class Order(BaseModel):
  item: str
  quantity: int

def process_order(node_input: Order) -> str:
  # If node_input is {'item': 'widget', 'quantity': 3},
  # it's auto-converted to Order(item='widget', quantity=3)
  return f"Order: {node_input.quantity}x {node_input.item}"

This works recursively for list[Model] and dict[str, Model] too.

Pydantic Schemas with LLM Agents (Recommended Pattern)

Use output_schema on LLM agents to get structured, JSON-serializable output. This avoids types.Content serialization issues and enables auto-conversion in downstream function nodes:

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

class ReviewResult(BaseModel):
  score: int
  feedback: str
  approved: bool

reviewer = LlmAgent(
    name="reviewer",
    model="gemini-2.5-flash",
    instruction="Review the code and provide structured feedback.",
    output_schema=ReviewResult,
)

# Downstream function node receives dict, auto-converted to Pydantic model
def process_review(node_input: ReviewResult) -> str:
  if node_input.approved:
    return f"Approved with score {node_input.score}"
  return f"Rejected: {node_input.feedback}"

Why use `output_schema`:

• LLM agent output becomes a dict (JSON-serializable) instead of types.Content
• Fixes TypeError when SQLite session service serializes JoinNode state
• Enables auto type conversion in downstream function nodes
• Provides structured data for programmatic access

Explicit FunctionNode

For more control, create a FunctionNode explicitly:

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

node = FunctionNode(
    my_func,
    name="custom_name",       # Override inferred name
    rerun_on_resume=True,     # Rerun after HITL interrupt
    retry_config=RetryConfig( # Retry on failure
        max_attempts=3,
        initial_delay=1.0,
    ),
)

@node Decorator

The @node decorator provides syntactic sugar:

from google.adk.workflow import node

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

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

# As a function call
my_node = node(some_func, name="renamed")

# Wrap as ParallelWorker
parallel = node(some_func, parallel_worker=True)

Prefer Typed Schemas Over Raw Dicts

Use Pydantic models for node inputs, outputs, and state instead of raw dict. This gives you validation, IDE autocomplete, and self-documenting code:

# ❌ Avoid: raw dicts are error-prone and opaque
def process(node_input: dict) -> dict:
  return {"status": "done", "count": node_input["items"]}

# ✅ Prefer: typed schemas
class TaskInput(BaseModel):
  items: list[str]
  priority: str = "normal"

class TaskResult(BaseModel):
  status: str
  count: int

def process(node_input: TaskInput) -> TaskResult:
  return TaskResult(status="done", count=len(node_input.items))

This applies to:

• Function node inputs/outputs: Use Pydantic models as node_input type hints and return types
• LLM agent `output_schema`: Always set output_schema=MyModel to get structured dict output instead of types.Content
• `RequestInput.response_schema`: Pass a Pydantic BaseModel class directly (e.g., response_schema=MyModel)
• State values: Store Pydantic model dicts (via .model_dump()) rather than hand-built dicts

FunctionNode auto-converts dict inputs to Pydantic models based on type hints (see Auto Type Conversion above), so typed schemas work seamlessly across the graph.

Emitting Content Events for Web UI Display

In the ADK web UI, only event.content is rendered to the user — event.output is internal and not displayed. When a function node produces user-facing output, yield a content event in addition to the output event:

from google.genai import types
from google.adk.events.event import Event

async def summarize(ctx: Context, node_input: str):
  result = f"Summary: {node_input}"

  # Content event: rendered in the web UI
  yield Event(content=types.ModelContent(result))

  # Output event: passed to downstream nodes
  yield Event(output=result)

LLM agents emit content events automatically. For function nodes that are terminal (no downstream edges) or produce user-visible intermediate results, add the content event so users see output in the web UI.

Events with Routes

Return an Event with a route for conditional branching:

def classify(node_input: str):
  if "urgent" in node_input:
    return Event(output=node_input, route="urgent")
  return Event(output=node_input, route="normal")

Events with State Updates

Update shared workflow state via the state constructor parameter:

def update_counter(node_input: str):
  return Event(
      output=node_input,
      state={"counter": 1, "last_input": node_input},
  )

Or use ctx.state directly:

def update_via_context(ctx: Context, node_input: str) -> str:
  ctx.state["counter"] = ctx.state.get("counter", 0) + 1
  return node_input

Type Validation (Important)

FunctionNode strictly type-checks node_input against the type hint. A TypeError is raised if the actual type doesn't match.

Union types: node_input: list | dict silently skips validation (FunctionNode detects Union via get_origin() and sets is_instance = True). This means Union hints won't crash, but they also won't catch wrong types — any value passes. Use isinstance checks inside the function body for actual validation.

Common pitfall: LLM agent -> function node. LlmAgentWrapper outputs types.Content (not str). If your function node follows an LLM agent and declares node_input: str, it will fail with:

TypeError: Parameter "node_input" expects type <class 'str'>
  but received type <class 'google.genai.types.Content'>

Fix: Use Any for node_input and extract text manually:

from typing import Any
from google.genai import types

def process(node_input: Any) -> str:
  # Handle types.Content from LLM agents
  if isinstance(node_input, types.Content):
    return ''.join(p.text for p in (node_input.parts or []) if p.text)
  return str(node_input) if node_input is not None else ''

Output type summary by predecessor:

Getting Started: Creating ADK Agents

Step-by-step guide covering environment setup, basic LLM agents, and workflow agents.

📋 New Agent Checklist

Use this checklist when creating a new agent to ensure it follows convention:

• [ ] Directory: Is there a directory for the agent?
• [ ] __init__.py: Does it contain from . import agent?
• [ ] agent.py: Does it define root_agent or app?
• [ ] .env: Is there a .env file with the appropriate API keys? (Do not commit to git)

💡 Quick Reference (CLI Commands)

• Create: adk create <agent_name> (Scaffolds a new agent project)
• Web UI: adk web <path_to_agent_dir> (Starts dev server at localhost:8000)
• Run CLI: adk run <path_to_agent_dir> (Interactive or query mode)

1. Set Up the Environment

Create a virtual environment and install the ADK:

# Create and activate virtual environment
python -m venv .venv
source .venv/bin/activate  # macOS/Linux

# Install the ADK package
pip install google-adk

Or with uv:

uv venv --python "python3.11" ".venv"
source .venv/bin/activate
uv pip install google-adk

2. Configure API Keys

Google AI Studio (recommended for getting started)

Obtain an API key from Google AI Studio.

Create a .env file in the agent directory:

GOOGLE_GENAI_USE_ENTERPRISE=FALSE
GOOGLE_API_KEY=YOUR_API_KEY

Vertex AI

For production use with Google Cloud:

GOOGLE_GENAI_USE_ENTERPRISE=TRUE
GOOGLE_CLOUD_PROJECT=your-project-id
GOOGLE_CLOUD_LOCATION=us-central1

Run gcloud auth application-default login to authenticate.

Vertex AI Express Mode

Combines Vertex AI with API key authentication:

GOOGLE_GENAI_USE_ENTERPRISE=TRUE
GOOGLE_API_KEY=YOUR_EXPRESS_MODE_KEY

3. Agent Directory Structure

The ADK CLI discovers agents by directory convention. Each agent directory must have:

my_agent/
├── __init__.py    # Must import the agent module
├── agent.py       # Must define root_agent
└── .env           # API keys (not committed to git)

__init__.py

from . import agent

Or generate the project with the CLI:

adk create my_agent

4. Basic LLM Agent with Tools

Before building workflow agents, understand the basic LLM agent pattern. An LlmAgent (also aliased as Agent) connects an LLM to tools and instructions:

agent.py

from google.adk.agents.llm_agent import Agent

def get_weather(city: str) -> dict:
  """Returns the current weather for a specified city."""
  # In production, call a real weather API
  return {
      "status": "success",
      "city": city,
      "weather": "sunny",
      "temperature": "72F",
  }

def get_current_time(city: str) -> dict:
  """Returns the current time in a specified city."""
  import datetime
  return {
      "status": "success",
      "city": city,
      "time": datetime.datetime.now().strftime("%I:%M %p"),
  }

root_agent = Agent(
    model="gemini-2.5-flash",
    name="root_agent",
    description="An assistant that provides weather and time information.",
    instruction="""You are a helpful assistant.
Use the get_weather tool to look up weather and
get_current_time to check the time in any city.
Always be friendly and concise.""",
    tools=[get_weather, get_current_time],
)

Key concepts

• `model`: The LLM to use (e.g., "gemini-2.5-flash", "gemini-2.5-pro")
• `instruction`: System prompt guiding the agent's behavior
• `tools`: Python functions the LLM can call. The function name, docstring, and type hints are sent to the LLM as the tool schema
• `description`: Used when this agent is a sub-agent (for transfer routing)
• `output_key`: Store the agent's final text output in session state under this key

Tool function conventions

• Use clear function names and docstrings — the LLM sees these
• Type-hint all parameters — they define the tool's input schema
• Return a dict or str — the return value becomes the tool response

5. Run the Agent

Web UI (primary debugging tool)

adk web my_agent/

Open http://localhost:8000. Select the agent from the dropdown, type a message, and see events in the Events tab.

Note: adk web is for development only, not production.

CLI mode

adk run my_agent/

API server

adk api_server my_agent/

Programmatic execution

import asyncio
from google.adk.runners import InMemoryRunner
from google.genai import types

async def main():
  from my_agent import agent

  runner = InMemoryRunner(
      app_name="my_app",
      agent=agent.root_agent,
  )

  session = await runner.session_service.create_session(
      app_name="my_app", user_id="user1"
  )

  content = types.Content(
      role="user", parts=[types.Part.from_text(text="What's the weather in Paris?")]
  )

  async for event in runner.run_async(
      user_id="user1",
      session_id=session.id,
      new_message=content,
  ):
    if event.content and event.content.parts:
      if event.content.parts[0].text:
        print(f"{event.author}: {event.content.parts[0].text}")

asyncio.run(main())

6. From LLM Agent to Workflow Agent

A Workflow extends the basic agent pattern with graph-based execution. Instead of a single LLM deciding what to do, define explicit nodes and edges:

agent.py — Minimal Workflow

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

5. Sample: Sequential Pipeline with LLM Agents

A code write-review-refactor pipeline using SequentialAgent:

agent.py

from google.adk.agents.llm_agent import LlmAgent
from google.adk.agents.sequential_agent import SequentialAgent

code_writer_agent = LlmAgent(
    name="CodeWriterAgent",
    model="gemini-2.5-flash",
    instruction="""You are a Python Code Generator.
Based *only* on the user's request, write Python code that fulfills the requirement.
Output *only* the complete Python code block.
""",
    description="Writes initial Python code based on a specification.",
    output_key="generated_code",
)

code_reviewer_agent = LlmAgent(
    name="CodeReviewerAgent",
    model="gemini-2.5-flash",
    instruction="""You are an expert Python Code Reviewer.
Review the following code:

{generated_code}

Provide feedback as a concise, bulleted list.
If the code is excellent, state: "No major issues found."
""",
    description="Reviews code and provides feedback.",
    output_key="review_comments",
)

code_refactorer_agent = LlmAgent(
    name="CodeRefactorerAgent",
    model="gemini-2.5-flash",
    instruction="""You are a Python Code Refactoring AI.
Improve the code based on the review comments.

**Original Code:**

{generated_code}

**Review Comments:**
{review_comments}

If no issues found, return the original code unchanged.
Output *only* the final Python code block.
""",
    description="Refactors code based on review comments.",
    output_key="refactored_code",
)

root_agent = SequentialAgent(
    name="CodePipelineAgent",
    sub_agents=[code_writer_agent, code_reviewer_agent, code_refactorer_agent],
    description="Executes a sequence of code writing, reviewing, and refactoring.",
)

Key patterns in this sample

• `output_key`: Each agent stores its output in session state, making it available to later agents
• `{generated_code}`: Instruction placeholders are resolved from session state at runtime
• `SequentialAgent`: Convenience wrapper that auto-generates START -> agent1 -> agent2 -> agent3 edges

6. Sample: Graph Workflow with Functions and Routing

A data processing pipeline with conditional routing:

agent.py

from google.adk.workflow import Workflow
from google.adk.events.event import Event
from google.adk.agents.context import Context

def parse_input(node_input: str) -> dict:
  """Parse the user's input into a structured format."""
  words = node_input.strip().split()
  return {"text": node_input, "word_count": len(words)}

def classify(node_input: dict):
  """Route based on input length."""
  if node_input["word_count"] > 10:
    return Event(output=node_input, route="long")
  return Event(output=node_input, route="short")

def handle_short(node_input: dict) -> str:
  return f"Short input ({node_input['word_count']} words): {node_input['text']}"

def handle_long(node_input: dict) -> str:
  return f"Long input ({node_input['word_count']} words). Summary: {node_input['text'][:50]}..."

root_agent = Workflow(
    name="classifier_workflow",
    input_schema=str,
    edges=[
        ('START', parse_input),
        (parse_input, classify),
        (classify, handle_short, "short"),
        (classify, handle_long, "long"),
    ],
)

7. Sample: Parallel Processing

Process a list of items concurrently:

agent.py

from google.adk.workflow import Workflow
from google.adk.workflow import node

def split_input(node_input: str) -> list:
  """Split comma-separated input into a list."""
  return [item.strip() for item in node_input.split(",")]

@node(parallel_worker=True)
def process_item(node_input: str) -> dict:
  """Process a single item (runs in parallel for each list item)."""
  return {"item": node_input, "length": len(node_input), "upper": node_input.upper()}

def format_results(node_input: list) -> str:
  """Format the parallel results into a readable summary."""
  lines = [f"- {r['item']}: {r['length']} chars -> {r['upper']}" for r in node_input]
  return "Results:\n" + "\n".join(lines)

root_agent = Workflow(
    name="parallel_processor",
    input_schema=str,
    edges=[
        ('START', split_input),
        (split_input, process_item),
        (process_item, format_results),
    ],
)

8. Sample: Workflow with LLM Agent and Tools

Combine function nodes with an LLM agent that has tools:

agent.py

from google.adk.agents.llm_agent import LlmAgent
from google.adk.workflow import Workflow
from google.adk.agents.context import Context

def get_weather(city: str) -> dict:
  """Get the current weather for a city."""
  # In production, call a real API
  return {"city": city, "temp": "72F", "condition": "sunny"}

def extract_city(node_input: str) -> str:
  """Extract city name from user input."""
  # Simple extraction; in production, use NLP or LLM
  return node_input.strip()

weather_agent = LlmAgent(
    name="weather_reporter",
    model="gemini-2.5-flash",
    instruction="""You are a friendly weather reporter.
Use the get_weather tool to look up the weather, then give
a natural-language weather report for the city.""",
    tools=[get_weather],
)

def format_output(ctx: Context, node_input: str) -> str:
  """Add a friendly sign-off."""
  return f"{node_input}\n\nHave a great day!"

root_agent = Workflow(
    name="weather_workflow",
    input_schema=str,
    edges=[
        ('START', extract_city),
        (extract_city, weather_agent),
        (weather_agent, format_output),
    ],
)

Troubleshooting

"No module named 'google.adk'"

Ensure the virtual environment is activated and google-adk is installed.

Agent not showing in adk web

Check that __init__.py contains from . import agent and agent.py defines root_agent.

API key errors

Verify .env is in the agent directory (not the parent) and contains a valid GOOGLE_API_KEY.

Model not found

Check the model name. Common models: gemini-2.5-flash, gemini-2.5-pro. The ADK also supports non-Google models (Anthropic, LiteLLM) with extra dependencies.

Human-in-the-Loop (HITL) Reference

Pause workflow execution to request user input and resume with their response.

📋 Agent Verification Checklist (HITL)

Use this checklist when implementing human-in-the-loop logic:

• [ ] Unique ID: Is the interrupt_id unique per iteration in loops? (Critical to prevent infinite loops)
• [ ] Resumability: For multi-step HITL, did you export an App with is_resumable=True?
• [ ] Resume Inputs: If rerun_on_resume=True (default for LLM nodes), does the node handle ctx.resume_inputs?

💡 Quick Reference

• Request Input: yield RequestInput(message="Question", response_schema=Schema)
• Resumable Config: ResumabilityConfig(is_resumable=True)

HITL works in two modes:

Resumable mode (recommended for multi-step HITL)

Export an App with resumability. The workflow checkpoints state and resumes at the interrupted node:

from google.adk.apps.app import App, ResumabilityConfig

app = App(
    name="my_app",
    root_agent=workflow_agent,
    resumability_config=ResumabilityConfig(is_resumable=True),
)

The agent loader checks for app before root_agent, so export both from agent.py.

Non-resumable mode (simpler, no App needed)

The workflow replays from START on each user response, reconstructing state from session events. No App or ResumabilityConfig needed — just define root_agent. This works for simple single-interrupt HITL but replays all nodes up to the interrupt point on each resume.

Imports

from google.adk.events.request_input import RequestInput
from google.adk.agents.context import Context
from google.adk.workflow import Workflow
from google.adk.apps.app import App, ResumabilityConfig

Basic Request Input

Yield or return a RequestInput to pause execution and ask the user for input:

# Yield from a generator
async def approval_gate(ctx: Context, node_input: str):
  yield RequestInput(
      message="Please approve this action:",
      response_schema={"type": "string"},
  )

# Or return directly from a regular function (no generator needed)
def evaluate_request(request: TimeOffRequest):
  if request.days <= 1:
    return TimeOffDecision(approved=True)  # Auto-approve
  return RequestInput(
      interrupt_id="manager_approval",
      message="Please review this time off request.",
      payload=request,
      response_schema=TimeOffDecision,
  )

The workflow pauses and emits a function call event to the user. When the user responds, the workflow resumes.

RequestInput Fields

from pydantic import BaseModel

class ApprovalResponse(BaseModel):
  approved: bool
  comment: str

RequestInput(
    interrupt_id="custom_id",         # Auto-generated UUID if omitted
    message="Question for user",      # Display message
    payload={"key": "value"},         # Custom data to include
    response_schema=ApprovalResponse, # Pydantic class, Python type, or JSON schema dict
)

Resume Behavior: rerun_on_resume

When a node is interrupted and the user responds, the rerun_on_resume flag controls what happens:

rerun_on_resume=False (default for FunctionNode)

The user's response becomes the node's output. The node is NOT re-executed:

from google.adk.workflow import FunctionNode

async def ask_approval(ctx: Context, node_input: str):
  yield RequestInput(message="Approve?")

# Node won't rerun; user's response is passed as output to next node
approval_node = FunctionNode(ask_approval, rerun_on_resume=False)

rerun_on_resume=True (default for LlmAgentWrapper)

The node is re-executed with the user's response available in ctx.resume_inputs:

async def interactive_node(ctx: Context, node_input: str):
  if ctx.resume_inputs:
    # Second run: user responded
    user_answer = list(ctx.resume_inputs.values())[0]
    yield Event(output=f"User said: {user_answer}")
  else:
    # First run: ask the user
    yield RequestInput(message="What should I do?")

HITL with LLM Agents

LLM agents support HITL via LongRunningFunctionTool:

from google.adk.tools.long_running_tool import LongRunningFunctionTool

def approval_tool(request: str) -> str:
  """Request human approval for an action."""
  return f"Approved: {request}"

llm_agent = LlmAgent(
    name="agent_with_approval",
    model="gemini-2.5-flash",
    instruction="When you need approval, use the approval_tool.",
    tools=[LongRunningFunctionTool(func=approval_tool)],
)

# LlmAgentWrapper has rerun_on_resume=True by default
agent = Workflow(
    name="hitl_workflow",
    edges=[
        ('START', llm_agent),
        (llm_agent, next_step),
    ],
)

Multi-Step HITL

A node can request input multiple times by checking ctx.resume_inputs:

async def multi_step_form(ctx: Context, node_input: str):
  if not ctx.resume_inputs:
    # Step 1: Ask for name
    yield RequestInput(
        interrupt_id="ask_name",
        message="What is your name?",
    )
    return

  if "ask_name" in ctx.resume_inputs and "ask_email" not in ctx.resume_inputs:
    # Step 2: Ask for email
    yield RequestInput(
        interrupt_id="ask_email",
        message="What is your email?",
    )
    return

  # All inputs collected
  name = ctx.resume_inputs["ask_name"]
  email = ctx.resume_inputs["ask_email"]
  yield Event(output={"name": name, "email": email})

HITL in Loops (Unique interrupt_id)

When a HITL node can fire multiple times in a loop (e.g. reject → revise → re-approve), you must use a unique `interrupt_id` per iteration. Reusing the same ID causes event-based state reconstruction to confuse earlier responses with the current interrupt, resulting in an infinite restart loop.

async def review(ctx: Context, node_input: Any):
  # Counter-based unique ID per review cycle
  review_count = ctx.state.get('review_count', 0)
  interrupt_id = f'review_{review_count}'

  response = ctx.resume_inputs.get(interrupt_id)
  if response:
    route = 'approved' if response.get('approved') else 'rejected'
    yield Event(
        output=response,
        route=route,
        state={'review_count': review_count + 1},
    )
    return

  yield RequestInput(
      interrupt_id=interrupt_id,
      message="Approve this plan?",
      response_schema=ApprovalSchema,
  )

Key points:

• Store a counter in ctx.state and increment on each response
• Use the counter in the interrupt_id (e.g. review_0, review_1, ...)
• Look up ctx.resume_inputs with the same counter-based ID
• This applies to both resumable and non-resumable modes

Resumability Configuration

Resumable mode (recommended for multi-step HITL)

from google.adk.apps.app import App, ResumabilityConfig

# Export BOTH root_agent and app from agent.py
root_agent = Workflow(name="my_workflow", edges=[...])

app = App(
    name="my_app",
    root_agent=root_agent,
    resumability_config=ResumabilityConfig(is_resumable=True),
)

When is_resumable=True:

• Workflow state is checkpointed in session's agent_states map
• On resume, the workflow loads checkpointed state and resumes at the interrupted node
• Required for multi-step HITL, LongRunningFunctionTool, and complex workflows

Non-resumable mode (simpler)

When is_resumable=False (default) or no App is exported:

• No state checkpointing — the workflow replays from START on each user response
• State is reconstructed from session events during replay
• Completed nodes are skipped; execution resumes at the interrupted node
• Works for simple single-interrupt HITL without needing App or ResumabilityConfig
• For multi-step HITL or complex workflows, use resumable mode instead

Responding to HITL Requests

From the client side, respond to function calls:

from google.genai import types

# Extract function_call_id from the interrupt event
function_call_id = interrupt_event.content.parts[0].function_call.id

# Create response
response = types.Content(
    role="user",
    parts=[types.Part(
        function_response=types.FunctionResponse(
            id=function_call_id,
            name="adk_request_input",
            response={"result": "User's answer here"},
        )
    )],
)

# Send response to resume the workflow
async for event in runner.run_async(
    user_id=user_id,
    session_id=session_id,
    new_message=response,
):
  # Process resumed workflow events
  pass

ADK Import Paths Quick Reference

📋 Agent Verification Checklist (Imports)

Use this checklist to ensure you are using the most idiomatic import paths:

• [ ] Canonical Imports: Did you use the short canonical imports where available (e.g., from google.adk import Agent) instead of the verbose ones?
• [ ] Avoid Deprecated: Are you avoiding deprecated paths (e.g., use McpToolset instead of MCPToolset)?

Canonical Imports (preferred, used by all samples)

from google.adk import Agent, Context, Event, Workflow
from google.adk.events import RequestInput
from google.adk.workflow import node, RetryConfig, Edge, JoinNode

Core Agents

Workflow Agents (Experimental)

Workflow Nodes

: : FunctionNode : | _LlmAgentWrapper (private, | from | : auto-used) : google.adk.workflow._llm_agent_wrapper : : : import _LlmAgentWrapper : | AgentNode | from google.adk.workflow._agent_node | : : import AgentNode : | _ToolNode (private) | from google.adk.workflow._tool_node | : : import _ToolNode : | JoinNode | from google.adk.workflow import | : : JoinNode : | Parallel-worker behavior (no public | Set parallel_worker=True on @node | : class) : or LlmAgent; the framework wraps : : : with an internal _ParallelWorker : | BaseNode, START | from google.adk.workflow import | : : BaseNode, START : | @node decorator | from google.adk.workflow import node` |

Workflow Events and Context

Task Mode

Tools

Built-in Tools

Runner and Session

App and Plugins

Models

Callbacks

Code Executors

Google GenAI Types

LLM Agent Nodes Reference

Embed LLM-powered agents as nodes in workflow graphs.

📋 Agent Verification Checklist (LLM Nodes)

Use this checklist to verify your LLM agent configuration:

• [ ] Output Type: If no output_schema is set, downstream now receives str (auto-extracted from types.Content). You can safely type-hint node_input: str.
• [ ] State Serialization: If this agent feeds into a JoinNode, did you set output_schema to avoid non-serializable types.Content errors?
• [ ] Instructions: Are {var} templates used in instructions resolving ONLY from ctx.state? (Not node_input)
• [ ] Config: Are instructions, tools, and response schema set on the LlmAgent directly, and NOT in generate_content_config?

💡 Quick Reference

• Chat Mode: Default. Multi-turn, keeps session history.
• Single-Turn Mode: Isolated. Set mode="single_turn" or rely on auto-wrapping defaults.
• Task Mode: Multi-turn within a task. Set mode="task".
• Stateless: Set include_contents="none" to ignore session history.

Imports

from google.adk.agents.llm_agent import LlmAgent
from google.adk.workflow._llm_agent_wrapper import _LlmAgentWrapper  # private
from google.adk.workflow import Workflow

Choosing the Right LLM Agent

Use `google.adk.agents.llm_agent.LlmAgent` in workflow edges. It is auto-wrapped as LlmAgentWrapper, which emits Event(output=...) for downstream data passing. This is required for any LLM agent that needs to pass output to downstream function nodes via node_input.

from google.adk.agents.llm_agent import LlmAgent

writer = LlmAgent(
    name="writer",
    model="gemini-2.5-flash",
    instruction="Write a short story.",
    output_schema=Story,
)

# writer is auto-wrapped as _LlmAgentWrapper — downstream gets Event(output=...)
agent = Workflow(
    name="pipeline",
    edges=[('START', writer), (writer, process_story)],
)

Basic LLM Node

from google.adk.agents.llm_agent import LlmAgent

writer = LlmAgent(
    name="writer",
    model="gemini-2.5-flash",
    instruction="Write a short story based on the user's prompt.",
)

reviewer = LlmAgent(
    name="reviewer",
    model="gemini-2.5-flash",
    instruction="Review the following story and provide feedback.",
)

agent = Workflow(
    name="story_pipeline",
    edges=[
        ('START', writer),      # Auto-wrapped as LlmAgentWrapper
        (writer, reviewer),
    ],
)

LLM Agent Output Types

When an LlmAgent runs as a workflow node, process_llm_agent_output (in _llm_agent_wrapper.py) sets event.output to:

• The concatenated text of the model's response (a str) — when

output_schema is not set.

• The validated dict (model_dump() of the Pydantic model) — when

output_schema=MyModel is set.

A downstream function node typed node_input: str therefore works in the default case, and node_input: dict works when output_schema is set.

Observability caveat: the value above is set on the event internally and forwarded to the next node, but event.output is `None` when you observe it from runner.run_async(...) for the LLM agent's own event — the framework clears it before the event reaches user code. Don't write tests that assert on event.output for an LLM agent's event; assert on the downstream node's output, on session.state[output_key], or on event.content.parts[*].text instead.

from pydantic import BaseModel

class CodeOutput(BaseModel):
  code: str
  language: str

writer = LlmAgent(
    name="writer",
    model="gemini-2.5-flash",
    instruction="Write code. Return JSON with 'code' and 'language' fields.",
    output_schema=CodeOutput,
)

# Downstream node receives a dict: {"code": "...", "language": "python"}
def process_code(node_input: dict) -> str:
  return node_input["code"]

Summary of LLM agent node output types:

Prefer `output_schema` when downstream nodes need structured access. Strings are fine for pass-through text, but a typed dict is easier to consume and is required when the predecessor feeds a JoinNode whose results land in a persistent session service (raw text is fine; objects that aren't JSON-serializable break DatabaseSessionService).

Auto-Wrapping Behavior

When you place an LlmAgent in workflow edges, it is auto-wrapped as _LlmAgentWrapper. The wrapper:

• Defaults to single_turn mode (agent sees only current input, not session history)
• Sets rerun_on_resume=True (reruns after HITL interrupts)
• Creates a content branch for isolation between parallel LLM agents

The mode is set on the LlmAgent itself, not the wrapper:

from google.adk.agents.llm_agent import LlmAgent

# single_turn (default when auto-wrapped): isolated, no session history
classifier = LlmAgent(
    name="classifier",
    model="gemini-2.5-flash",
    instruction="Classify the input as positive, negative, or neutral.",
    output_schema=ClassificationResult,
)

# task mode: supports HITL, multi-turn within the task
task_agent = LlmAgent(
    name="task_agent",
    model="gemini-2.5-flash",
    mode="task",
    instruction="Process the request.",
)

LlmAgent Configuration

Instructions

Dynamic instructions with placeholders resolved from session state. `{var}` templates only resolve from `ctx.state` — `node_input` is NOT available in templates. To use predecessor data in instructions, store it in state first (via Event(state={...}) or output_key):

agent = LlmAgent(
    name="personalized",
    model="gemini-2.5-flash",
    instruction="""You are helping {user_name}.
Their preferences are: {preferences}.
Respond in {language}.""",
)
# {user_name}, {preferences}, {language} resolved from session state
# Missing variables raise KeyError at runtime — use {var?} for optional:
# instruction="Current mood: {mood?}"  # empty string if 'mood' not in state

Template variable behavior:

Instruction provider function for fully dynamic instructions:

from google.adk.agents.readonly_context import ReadonlyContext

def build_instruction(ctx: ReadonlyContext) -> str:
  agents = ctx.state.get("active_agents", [])
  return f"Coordinate these agents: {', '.join(agents)}"

agent = LlmAgent(
    name="coordinator",
    model="gemini-2.5-flash",
    instruction=build_instruction,
)

Output Schema

Structure LLM output with Pydantic models:

from pydantic import BaseModel

class ReviewResult(BaseModel):
  score: int
  feedback: str
  approved: bool

reviewer = LlmAgent(
    name="reviewer",
    model="gemini-2.5-flash",
    instruction="Review the code and provide structured feedback.",
    output_schema=ReviewResult,
)

When used as a workflow node, the output becomes a dict (via model_dump()) as node_input for the next node.

Output Key

Store agent output in session state:

agent = LlmAgent(
    name="writer",
    model="gemini-2.5-flash",
    instruction="Write a draft.",
    output_key="draft",  # Stores output in state['draft']
)

include_contents

Control conversation history:

agent = LlmAgent(
    name="stateless",
    model="gemini-2.5-flash",
    instruction="Process this input independently.",
    include_contents="none",  # Don't include session history
)

Tools

Add tools to LLM agents:

def search_database(query: str) -> str:
  """Search the database for relevant records."""
  return f"Results for: {query}"

def send_email(to: str, subject: str, body: str) -> str:
  """Send an email to the specified address."""
  return "Email sent"

agent = LlmAgent(
    name="assistant",
    model="gemini-2.5-flash",
    instruction="Help the user with their request.",
    tools=[search_database, send_email],
)

Tools can be:

• Python functions (auto-wrapped as FunctionTool)
• BaseTool instances
• BaseToolset instances (e.g., MCP toolsets)

Callbacks

Before Model Callback

Intercept or modify LLM requests. Return an LlmResponse to skip the LLM call; return None to proceed:

from google.adk.agents.callback_context import CallbackContext
from google.adk.models.llm_request import LlmRequest
from google.adk.models.llm_response import LlmResponse

def guard_callback(
    callback_context: CallbackContext,
    llm_request: LlmRequest,
) -> LlmResponse | None:
  for content in llm_request.contents:
    if content.parts:
      for part in content.parts:
        if part.text and "unsafe" in part.text:
          return LlmResponse(
              content=types.ModelContent("I cannot process that.")
          )
  return None  # Proceed with normal LLM call

agent = LlmAgent(
    name="guarded",
    model="gemini-2.5-flash",
    before_model_callback=guard_callback,
)

After Model Callback

Transform LLM responses. Return an LlmResponse to replace; return None to keep original:

def log_response(
    callback_context: CallbackContext,
    llm_response: LlmResponse,
) -> LlmResponse | None:
  print(f"LLM responded: {llm_response.content}")
  return None  # Use original response

agent = LlmAgent(
    name="logged",
    model="gemini-2.5-flash",
    after_model_callback=log_response,
)

Before/After Tool Callbacks

Intercept tool calls. Return a dict to use as tool response (skipping actual execution); return None to proceed:

from google.adk.tools.base_tool import BaseTool
from google.adk.tools.tool_context import ToolContext

def audit_tool(
    tool: BaseTool,
    args: dict[str, Any],
    tool_context: ToolContext,
) -> dict | None:
  print(f"Calling tool {tool.name} with args: {args}")
  return None  # Proceed with tool call

def validate_tool_result(
    tool: BaseTool,
    args: dict[str, Any],
    tool_context: ToolContext,
    tool_response: dict,
) -> dict | None:
  if "error" in tool_response:
    return {"result": "Tool execution failed, please try again."}
  return None  # Use original result

agent = LlmAgent(
    name="audited",
    model="gemini-2.5-flash",
    tools=[my_tool],
    before_tool_callback=audit_tool,
    after_tool_callback=validate_tool_result,
)

Multiple Callbacks

Pass a list of callbacks. They execute in order until one returns non-None:

agent = LlmAgent(
    name="multi_callback",
    model="gemini-2.5-flash",
    before_model_callback=[safety_check, rate_limiter, logger],
)

Error Callbacks

Handle LLM or tool errors gracefully:

def handle_model_error(
    callback_context: CallbackContext,
    llm_request: LlmRequest,
    error: Exception,
) -> LlmResponse | None:
  return LlmResponse(
      content=types.ModelContent("Service temporarily unavailable.")
  )

def handle_tool_error(
    tool: BaseTool,
    args: dict[str, Any],
    tool_context: ToolContext,
    error: Exception,
) -> dict | None:
  return {"error": str(error), "fallback": True}

agent = LlmAgent(
    name="resilient",
    model="gemini-2.5-flash",
    on_model_error_callback=handle_model_error,
    on_tool_error_callback=handle_tool_error,
)

All Callback Types

All callbacks can be sync or async. All accept a single callback or a list.

Generate Content Config

Fine-tune LLM generation:

from google.genai import types

agent = LlmAgent(
    name="creative",
    model="gemini-2.5-flash",
    instruction="Write creative stories.",
    generate_content_config=types.GenerateContentConfig(
        temperature=0.9,
        top_p=0.95,
        max_output_tokens=2048,
    ),
)

Agent Transfer

Agents can transfer control to sub-agents:

specialist = LlmAgent(
    name="specialist",
    model="gemini-2.5-flash",
    instruction="Handle specialized requests.",
)

coordinator = LlmAgent(
    name="coordinator",
    model="gemini-2.5-flash",
    instruction="Route requests to the specialist when needed.",
    sub_agents=[specialist],
)

Control transfer behavior:

agent = LlmAgent(
    name="isolated",
    model="gemini-2.5-flash",
    disallow_transfer_to_parent=True,
    disallow_transfer_to_peers=True,
)

Multi-Agent Patterns

📋 Agent Verification Checklist (Multi-Agent)

Use this checklist when setting up multi-agent systems:

• [ ] Description: Does every sub-agent have a clear description? (Used by LLM for routing or tool generation)
• [ ] Model Inheritance: Did you let sub-agents inherit the model from the coordinator to avoid duplication?
• [ ] Loop Termination: If using LoopAgent, is there a clear way to call exit_loop to prevent infinite loops?

💡 Quick Reference

• Sequential: SequentialAgent(sub_agents=[a, b, c])
• Parallel: ParallelAgent(sub_agents=[a, b, c])
• Loop: LoopAgent(sub_agents=[a, b], max_iterations=5)

LLM-Based Multi-Agent (Chat Transfer)

from google.adk.agents.llm_agent import Agent

researcher = Agent(
    name='researcher',
    description='Researches topics.',
    instruction='You research topics and provide findings.',
    tools=[search_tool],
)

writer = Agent(
    name='writer',
    description='Writes content.',
    instruction='You write content based on research.',
)

root_agent = Agent(
    model='gemini-2.5-flash',
    name='coordinator',
    instruction=(
        'Delegate research to the researcher and '
        'writing to the writer.'
    ),
    sub_agents=[researcher, writer],
)

Key rules:

• Only the root agent needs model=. Sub-agents inherit it.
• Each sub-agent needs a description (used for routing).
• Transfer between agents is automatic via LLM reasoning.
• disallow_transfer_to_parent=True prevents back-transfer.
• disallow_transfer_to_peers=True prevents peer-transfer.

Task-Based Multi-Agent (Structured Delegation)

For structured input/output, use task mode instead of chat transfer. See `task-mode.md` for full details.

from google.adk import Agent

worker = Agent(
    name='worker',
    mode='task',                     # or 'single_turn'
    input_schema=WorkerInput,
    output_schema=WorkerOutput,
    instruction='Do work, then call finish_task.',
    description='Performs structured work.',
)

root_agent = Agent(
    name='coordinator',
    model='gemini-2.5-flash',
    sub_agents=[worker],
    instruction='Delegate to worker via request_task_worker.',
)

Non-LLM Orchestration Agents

SequentialAgent

Runs sub-agents in order, one after another:

from google.adk.agents.sequential_agent import SequentialAgent

root_agent = SequentialAgent(
    name='pipeline',
    sub_agents=[step1_agent, step2_agent, step3_agent],
)

ParallelAgent

Runs sub-agents concurrently:

from google.adk.agents.parallel_agent import ParallelAgent

root_agent = ParallelAgent(
    name='fan_out',
    sub_agents=[task_a, task_b, task_c],
)

LoopAgent

Repeats sub-agents until exit_loop is called:

from google.adk.tools import exit_loop
from google.adk.agents.loop_agent import LoopAgent

looping_agent = Agent(
    name='checker',
    tools=[exit_loop],
    instruction='Check the result and call exit_loop if done.',
)

root_agent = LoopAgent(
    name='retry_loop',
    sub_agents=[worker_agent, looping_agent],
    max_iterations=5,
)

Model Configuration

• Default model: gemini-2.5-flash
• Override globally: Agent.set_default_model('gemini-2.5-pro')
• Model inheritance: sub-agents inherit parent's model if not set
• Non-Gemini models via LiteLlm:

  from google.adk.models.lite_llm import LiteLlm
  root_agent = Agent(model=LiteLlm(model='anthropic/claude-sonnet-4-20250514'), ...)

Common Pitfalls

• Agent stuck in sub-agent: Sub-agent has no path back to parent.

Set disallow_transfer_to_parent=False (default) or add explicit transfer instructions.

• Wrong agent handles request: Ambiguous description fields. Make

each agent's description clearly differentiate its scope.

• Circular imports: Define all agents in a single agent.py file,

or use a shared module for sub-agents.

Parallel Execution, Fan-Out, and Fan-In Reference

Execute multiple nodes concurrently and collect their results.

📋 Agent Verification Checklist (Parallel & Fan-Out)

Use this checklist when implementing parallel patterns:

• [ ] JoinNode Serialization: If LLM agents feed into a JoinNode, did you set output_schema on them to prevent JSON serialization errors?
• [ ] ParallelWorker Usage: Did you avoid using parallel_worker=True on fan-out nodes? (It expects a list input)
• [ ] Multi-Trigger vs Join: Do you understand that Multi-Trigger fires downstream once per branch, while JoinNode waits and fires once with merged dict?

💡 Quick Reference

• Fan-Out (Tuple): ('START', (node_a, node_b))
• Fan-In (JoinNode): ((node_a, node_b), join_node)
• List Worker: @node(parallel_worker=True) (Takes list, outputs list)

Imports

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

Parallel-worker behavior is opted into via the parallel_worker=True flag on @node or LlmAgent. The underlying wrapper class is internal — don't import it directly.

Fan-Out: Multiple Branches

Send output to multiple nodes simultaneously using tuple syntax:

def analyze_text(node_input: str) -> str:
  return f"Analysis: {node_input}"

def translate_text(node_input: str) -> str:
  return f"Translation: {node_input}"

def summarize_text(node_input: str) -> str:
  return f"Summary: {node_input}"

agent = Workflow(
    name="fan_out",
    edges=[
        ('START', (analyze_text, translate_text, summarize_text)),
    ],
)

Each branch receives the same input and runs concurrently.

Fan-In: JoinNode

Collect outputs from multiple branches before continuing:

join = JoinNode(name="collect_results")

agent = Workflow(
    name="fan_out_fan_in",
    edges=[
        ('START', (analyze_text, translate_text, summarize_text)),
        ((analyze_text, translate_text, summarize_text), join),
        (join, final_processor),
    ],
)

JoinNode Output Format

JoinNode outputs a dictionary mapping predecessor names to their outputs:

# JoinNode output:
# {
#   "analyze_text": "Analysis: hello",
#   "translate_text": "Translation: hello",
#   "summarize_text": "Summary: hello",
# }

def final_processor(node_input: dict) -> str:
  analysis = node_input["analyze_text"]
  translation = node_input["translate_text"]
  summary = node_input["summarize_text"]
  return f"Combined: {analysis}, {translation}, {summary}"

JoinNode Behavior

• Waits for all predecessor nodes to complete
• Emits intermediate events while still waiting (downstream not triggered until all inputs received)
• Only triggers downstream when all inputs are received
• Stores partial inputs in workflow state

Serialization warning: JoinNode stores partial inputs in session state while waiting. If predecessors are LLM agents without output_schema, the stored values are types.Content objects which are not JSON-serializable. This causes TypeError with SQLite/database session services. Fix: use output_schema on LLM agents feeding into a JoinNode.

Parallel workers: process lists in parallel

Apply the same node to each item in a list concurrently by setting the parallel_worker=True flag. The framework wraps the node internally — there is no public ParallelWorker class to import.

from google.adk.workflow import node, Workflow

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

def produce_list(node_input: str) -> list:
  return [1, 2, 3, 4, 5]

agent = Workflow(
    name="parallel_processing",
    edges=[
        ('START', produce_list),
        (produce_list, process_item),
    ],
)
# Output: [2, 4, 6, 8, 10]

Behavior

• Input: a list (or single item, which gets wrapped in a list)
• Output: a list of results in the same order as inputs
• Empty list input produces empty list output
• Each item is processed by a dynamically created worker node
• Default rerun_on_resume=True

Parallel workers with Agents

Set parallel_worker=True directly on an Agent — no extra wrapping needed:

from google.adk import Agent

explain_topic = Agent(
    name="explain_topic",
    instruction="Explain how this topic relates to the original topic: \"{topic}\".",
    output_schema=TopicExplanation,
    parallel_worker=True,  # Each list item processed by a cloned agent
)

agent = Workflow(
    name="parallel_analysis",
    edges=[
        ('START', process_input, find_related_topics, explain_topic, aggregate),
    ],
)

Do NOT use `parallel_worker=True` on fan-out nodes. Fan-out edges (a, (b, c, d)) already run nodes in parallel. Adding parallel_worker=True makes the node expect a list input and iterate over it — if it receives a single value or None, it produces no output and the JoinNode gets nothing.

Multi-Trigger (Fan-Out to Shared Downstream)

Fan-out branches that all feed a single downstream node. The downstream node is triggered once per branch:

async def send_message(node_input: Any):
  yield Event(message=f"Triggered for input: {node_input}")

agent = Workflow(
    name="root_agent",
    edges=[(
        "START",
        (make_uppercase, count_characters, reverse_string),
        send_message,
    )],
    input_schema=str,
)

This differs from JoinNode: here send_message fires 3 times (once per branch), while JoinNode waits for all branches and fires once with a merged dict.

Diamond Pattern

Fan-out then fan-in (diamond shape):

def splitter(node_input: str) -> str:
  return node_input

def branch_a(node_input: str) -> str:
  return f"A: {node_input}"

def branch_b(node_input: str) -> str:
  return f"B: {node_input}"

join = JoinNode(name="merge")

def combiner(node_input: dict) -> str:
  return f"Combined: {node_input['branch_a']} + {node_input['branch_b']}"

agent = Workflow(
    name="diamond",
    edges=[
        ('START', splitter),
        (splitter, (branch_a, branch_b)),
        ((branch_a, branch_b), join),
        (join, combiner),
    ],
)

SequentialAgent and ParallelAgent

Convenience subclasses for common patterns:

from google.adk.agents.sequential_agent import SequentialAgent
from google.adk.agents.parallel_agent import ParallelAgent

# Sequential: runs sub_agents in order
pipeline = SequentialAgent(
    name="pipeline",
    sub_agents=[writer_agent, reviewer_agent, editor_agent],
)
# Equivalent to: START -> writer -> reviewer -> editor

# Parallel: runs sub_agents concurrently
parallel = ParallelAgent(
    name="concurrent",
    sub_agents=[analyzer_agent, translator_agent, summarizer_agent],
)
# Equivalent to: START -> (analyzer, translator, summarizer)

Routing and Conditional Branching Reference

Route workflow execution along different paths based on node outputs.

📋 Agent Verification Checklist (Routing)

Use this checklist when implementing routing logic:

• [ ] Syntax: Is the preferred dict syntax used for mapping routes to targets? (Avoid verbose individual edges)
• [ ] Loops: Are cycles (loops) routed? (Unconditional cycles are rejected during validation)
• [ ] Triggering: If a node has conditional routing, do ALL outgoing edges have routes? (To avoid unintended triggering by unconditional edges)

💡 Quick Reference

• Dict Routing: (source_node, {"route_a": target_a, "route_b": target_b})
• Sequence: ("START", step_a, step_b, step_c)
• Default: "__DEFAULT__" (Fallback route)

Basic Routing

A node emits an Event with a route value. Use dict syntax to map routes to target nodes:

from google.adk import Event, Workflow

def classify(node_input: str):
  if "error" in node_input:
    return Event(output=node_input, route="error")
  return Event(output=node_input, route="success")

def handle_success(node_input: str) -> str:
  return f"Success: {node_input}"

def handle_error(node_input: str) -> str:
  return f"Error: {node_input}"

agent = Workflow(
    name="router",
    edges=[
        ('START', classify),
        (classify, {"success": handle_success, "error": handle_error}),
    ],
)

Routing Map (Dict Syntax) — Preferred

The dict syntax is the idiomatic way to express routing. It maps route values to target nodes in a single edge tuple:

edges = [
    ("START", process_input, classifier, route_on_category),
    (route_on_category, {
        "question": answer_question,
        "statement": comment_on_statement,
        "other": handle_other,
    }),
]

This replaces verbose individual routed edges:

# ❌ Verbose — avoid
(classifier, answer_question, "question"),
(classifier, comment_on_statement, "statement"),
(classifier, handle_other, "other"),

# ✅ Preferred — dict syntax
(classifier, {"question": answer_question, "statement": comment_on_statement, "other": handle_other}),

Sequence Shorthand (Tuple Chains)

A tuple with more than 2 elements creates a sequential chain:

# Shorthand: tuple creates chain edges
edges = [("START", step_a, step_b, step_c)]
# Equivalent to: [("START", step_a), (step_a, step_b), (step_b, step_c)]

Combine with dict routing:

edges = [
    ("START", process_input, classify, route_on_result),
    (route_on_result, {"approved": send, "rejected": discard}),
]

Route Value Types

Routes can be str, bool, or int:

# String routes (most common)
(decision_node, {"approve": path_a, "reject": path_b})

# Boolean routes
(decision_node, {True: yes_path, False: no_path})

# Integer routes
(decision_node, {0: path_0, 1: path_1})

Default Route

Use '__DEFAULT__' as a fallback when no other route matches:

edges = [
    ("START", classify),
    (classify, {
        "success": handler_a,
        "error": handler_b,
        "__DEFAULT__": fallback_handler,
    }),
]

Only one default route per node is allowed.

No duplicate edges: Two edges from the same source to the same target are rejected, even with different routes. If you need both a named route and __DEFAULT__ to reach the same destination, use a thin wrapper function for the default path.

Dynamic Routing with Functions

A function node that emits different routes based on runtime data:

from google.adk import Context, Event

def route_on_score(ctx: Context, node_input: dict):
  score = node_input.get("score", 0)
  if score > 0.8:
    return Event(output=node_input, route="high")
  elif score > 0.5:
    return Event(output=node_input, route="medium")
  else:
    return Event(output=node_input, route="low")

agent = Workflow(
    name="scored_router",
    edges=[
        ("START", compute_score, route_on_score),
        (route_on_score, {
            "high": premium_handler,
            "medium": standard_handler,
            "low": basic_handler,
        }),
    ],
)

Multi-Route (Fan-Out via Route)

A node can output multiple routes to trigger multiple downstream paths simultaneously:

def fan_out_router(node_input: str):
  return Event(output=node_input, route=["path_a", "path_b"])

agent = Workflow(
    name="multi_route",
    edges=[
        ("START", fan_out_router),
        (fan_out_router, {"path_a": branch_a, "path_b": branch_b}),
    ],
)

List of Routes on a Single Edge

An edge can match multiple routes by passing a list as the route value. The edge fires if the node output matches any route in the list:

edges = [
    ("START", classifier),
    (classifier, {"route_z": handler_b}),
    # handler_a fires on either route_x or route_y
    (classifier, handler_a, ["route_x", "route_y"]),
]

This is useful when multiple route values should lead to the same downstream node without duplicating edges. Note: list-of-routes on a single edge uses the 3-tuple syntax since dict syntax maps one route to one target.

Self-Loop

A node can route back to itself:

def guess_number(target_number: int):
  guess = random.randint(0, 10)
  yield Event(message=f'Guessing {guess}...')
  if guess == target_number:
    yield Event(message='Correct!')
  else:
    yield Event(route='guessed_wrong')

agent = Workflow(
    name='root_agent',
    edges=[
        ('START', validate_input, guess_number),
        (guess_number, {'guessed_wrong': guess_number}),
    ],
)

Revision Loop

A common pattern: route back to an earlier node for revision, or forward for approval:

edges = [
    ("START", process_input, draft_email, human_review),
    (human_review, {
        "revise": draft_email,
        "approved": send,
        "rejected": discard,
    }),
]

Important: Cycles must have at least one routed edge (unconditional cycles are rejected during graph validation).

Unconditional Edges

Edges without a route value are unconditional — they always fire:

edges = [
    ('START', node_a),       # Unconditional
    (node_a, node_b),        # Unconditional (always fires)
]

Important: Unrouted edges always fire, regardless of whether the output event has a route. If a node has conditional routing, ALL outgoing edges should have routes to avoid unintended triggering.

Session, Memory, and Artifact Patterns

📋 Agent Verification Checklist (Session & State)

Use this checklist when managing state and artifacts:

• [ ] State Mutation: Did you use ctx.state['key'] = value instead of reassigning state = {...}?
• [ ] Instruction Placeholders: Did you use {var?} for variables that might not be in state yet?
• [ ] Key Collisions: In parallel workflows, do state keys have unique names or appropriate prefixes (e.g., app:) to prevent overwrites?

💡 Quick Reference (State Keys)

• Required: {key} in instructions (raises error if missing).
• Optional: {key?} in instructions (empty string if missing).
• App Scope: app:key (Shared across agents).
• Agent Scope: key (Default, scoped to current agent).

Session State

Session state is a dict that persists across turns within a session. Access via tool_context.state or instruction placeholders:

# In instruction (template variable substitution)
instruction = 'Current user: {user_name}'

# In tool
def my_tool(tool_context: ToolContext):
  tool_context.state['user_name'] = 'Alice'

# In callback
def before_agent(callback_context):
  callback_context.state['_time'] = datetime.now().isoformat()

State key conventions:

• app:key -- app-level state (shared across agents)
• key -- agent-level state (scoped to current agent)
• _key -- convention for internal/framework state
• {key?} in instruction -- optional placeholder (empty if missing)
• {key} in instruction -- required placeholder (error if missing)

Session Services

from google.adk import Runner
from google.adk.sessions import InMemorySessionService

runner = Runner(
    agent=root_agent,
    app_name='my_app',
    session_service=InMemorySessionService(),
)

Artifacts

Artifacts store non-textual data (files, images) associated with sessions:

from google.genai import types

# Save from tool
async def save_chart(tool_context: ToolContext):
  chart_bytes = generate_chart()
  part = types.Part.from_bytes(data=chart_bytes, mime_type='image/png')
  version = await tool_context.save_artifact('chart.png', part)

# Load from tool
async def get_chart(tool_context: ToolContext):
  part = await tool_context.load_artifact('chart.png')
  return part.inline_data.data

Memory Services

Long-term recall across sessions:

from google.adk.memory.in_memory_memory_service import InMemoryMemoryService

runner = Runner(
    agent=root_agent,
    memory_service=InMemoryMemoryService(),
    ...
)

Use load_memory and preload_memory tools to access memory from within agents.

Common Pitfalls

• State not persisting: Assigning to state instead of mutating.

Use tool_context.state['key'] = value (not state = {'key': value}).

• State overwritten by parallel tools: Multiple tools modifying same

key concurrently. Use unique keys per tool, or app: prefix for shared state.

State and Events Reference

Manage shared state across workflow nodes and understand the event system.

📋 Agent Verification Checklist (State & Events)

Use this checklist when working with state and events:

• [ ] State Updates: Did you use Event(state=...) for state updates? (Captures delta in event history)
• [ ] Parameter Resolution: Are custom parameters named after keys in ctx.state?
• [ ] Output Serialization: Is event.output JSON-serializable? (Required for DB session services)
• [ ] Web UI Display: Did you use Event(message=...) for output meant for users?

💡 Quick Reference (Resolution Order)

1. `ctx`: Workflow Context object. 2. `node_input`: Predecessor output. 3. Other names: Looked up from ctx.state[param_name].

Workflow Context

Every node receives a Context object (when declaring a ctx parameter):

from google.adk.agents.context import Context

def my_node(ctx: Context, node_input: str) -> str:
  # Access shared state
  value = ctx.state.get("key", "default")

  # Write to state
  ctx.state["key"] = "new_value"

  # Access session info
  session_id = ctx.session.id
  invocation_id = ctx.invocation_id

  # Get node metadata
  node_path = ctx.node_path          # e.g., "MyWorkflow/my_node"
  run_id = ctx.run_id                # this node-run's identifier
  attempt = ctx.attempt_count        # 1 on first attempt, ≥1 thereafter

  return f"Processed: {value}"

Context Properties

Common Properties (available everywhere)

Workflow-Only Properties

: : : "WorkflowA/node1") : | run_id | str | Identifier for this node-run (e.g., | : : : "1", "2") : | attempt_count | int | Retry attempt number (1 on first try) | | resume_inputs | dict[str, Any] | Inputs for resuming (keyed by | : : : interrupt_id) :

Workflow-Only Methods

State Management

State is shared across all nodes in a workflow invocation. Prefer `Event(state=...)` over `ctx.state[...] =` for setting state:

# ✅ Preferred: set state via Event (persisted in event history, replayable)
def node_a(node_input: str):
  return Event(
      output="done",
      state={"user_data": {"name": "Alice", "score": 95}},
  )

# ❌ Avoid: direct ctx.state mutation (not captured in event history)
def node_a(ctx: Context, node_input: str) -> str:
  ctx.state["user_data"] = {"name": "Alice", "score": 95}
  return "done"

Why `Event(state=...)` is preferred:

• State deltas are persisted in event history as event.actions.state_delta
• Non-resumable HITL can reconstruct state by replaying events
• Makes state changes explicit and traceable
• ctx.state mutations are side effects that may be lost on replay

Reading state is always done via ctx.state:

def node_b(ctx: Context, node_input: str) -> str:
  user = ctx.state["user_data"]
  return f"User {user['name']} scored {user['score']}"

The state dict is stored as event.actions.state_delta and applied to the session.

State as Function Parameters

FunctionNode automatically resolves parameters from state:

# If ctx.state["user_name"] = "Alice" and ctx.state["threshold"] = 0.5
def my_node(node_input: str, user_name: str, threshold: float) -> str:
  # user_name = "Alice" (from state)
  # threshold = 0.5 (from state)
  return f"{user_name}: {node_input} (threshold={threshold})"

Resolution order:

1. ctx -> Context object 2. node_input -> predecessor output 3. Other names -> ctx.state[param_name] (with auto type conversion) 4. Default values if not in state

Event Fields

Workflow Data Rules

• `Event.output` must be JSON-serializable. FunctionNode auto-converts Pydantic BaseModel returns via model_dump(), so returning a model is safe. But types.Content and other non-serializable objects will fail with SQLite/database session services.
• `output_key` stores dicts, not BaseModel instances. LLM agents with output_schema use validate_schema() → model_dump() internally, so ctx.state[output_key] is always a plain dict.
• `ctx.state.get(key)` returns a dict. Use dict access (data["field"]) or reconstruct the model (MyModel(**data)) if you need typed access.

# Reading output_key from state — it's a dict, not a BaseModel
def use_plan(ctx: Context, node_input: Any) -> str:
  plan = ctx.state.get('task_plan', {})  # dict, not TaskPlan
  return plan['project_name']            # dict access

  # Or reconstruct if you need typed access:
  plan_model = TaskPlan(**plan)
  return plan_model.project_name

Content Events (User-Visible Output)

In the ADK web UI, only event.content is rendered — event.output is internal and not displayed. Emit content events for any user-facing output:

# Simple text message
yield Event(message="Processing step 1...")

# Multimodal message (text + image)
from google.genai import types
yield Event(
    message=[
        types.Part.from_text(text="Here is the result:"),
        types.Part.from_bytes(data=image_bytes, mime_type="image/png"),
    ]
)

# Streaming: multiple messages from same node
async def verbose_node(ctx: Context, node_input: str):
  yield Event(message="Processing step 1...")
  await asyncio.sleep(1.0)
  yield Event(message="Processing step 2...")
  yield Event(output="final result")

Workflow Output

The Workflow emits its own output Event in _finalize_workflow after all nodes complete. Terminal nodes (nodes with no outgoing edges) have their data collected and emitted as the workflow's output. This output event has author=workflow.name and node_path=workflow's own path.

Task Mode: Structured Delegation

Delegate structured tasks to sub-agents with typed input/output schemas.

📋 Agent Verification Checklist (Task Mode)

Use this checklist to verify your Task Mode configuration:

• [ ] Mode Setting: Did you explicitly set mode='task' or mode='single_turn' on the sub-agent?
• [ ] Description: Does the sub-agent have a clear description? (Crucial for the auto-generated tool's description)
• [ ] Schemas: Are input_schema and output_schema defined as Pydantic models? (If not, defaults are used)
• [ ] Completion: Does the sub-agent know it must call finish_task to return results to the coordinator?

💡 Quick Reference (Generated Tools)

• `request_task_{agent_name}`: Generated on the coordinator to delegate tasks.
• `finish_task`: Generated on the sub-agent to return results and complete the task.

Overview

ADK agents support three delegation modes via the mode parameter on Agent:

Imports

from google.adk import Agent
from pydantic import BaseModel

Note: Task mode uses Agent (aliased from LlmAgent) from google.adk. Both task sub-agents and coordinators use the same Agent class — set mode='task' or mode='single_turn' on sub-agents.

Task Mode (mode='task')

A task agent receives structured input via request_task_{name}, can interact with the user for clarification, and returns structured output via finish_task.

Delegation Lifecycle

1. User asks the coordinator to do something 2. Coordinator calls request_task_{agent_name}(...) with structured input 3. Task agent receives the input, works on it (may use tools, may chat with user) 4. Task agent calls finish_task(...) with structured output 5. Coordinator receives the result and responds to the user

Example

from google.adk import Agent
from pydantic import BaseModel

class ResearchInput(BaseModel):
  topic: str
  depth: str = 'standard'

class ResearchOutput(BaseModel):
  summary: str
  key_findings: str
  confidence: str

def search_web(query: str) -> str:
  """Search the web for information."""
  return f'Results for "{query}": ...'

def analyze_sources(sources: str) -> str:
  """Analyze and synthesize source material."""
  return f'Analysis of {len(sources.split())} words complete.'

researcher = Agent(
    name='researcher',
    mode='task',
    input_schema=ResearchInput,
    output_schema=ResearchOutput,
    instruction=(
        'You are a research assistant. When given a topic:\n'
        '1. Use search_web to find information.\n'
        '2. Use analyze_sources to synthesize findings.\n'
        '3. If the user asks for changes, adjust your research.\n'
        '4. Call finish_task with summary, key_findings, and confidence.'
    ),
    description='Researches topics using web search and analysis.',
    tools=[search_web, analyze_sources],
)

root_agent = Agent(
    name='coordinator',
    model='gemini-2.5-flash',
    sub_agents=[researcher],
    instruction=(
        'When the user asks you to research something, delegate to'
        ' the researcher using request_task_researcher. After the'
        ' researcher completes, summarize the results for the user.'
    ),
)

Single-Turn Mode (mode='single_turn')

A single-turn agent completes autonomously with no user interaction. It receives input, does its work, and returns a result.

Example

class SummaryOutput(BaseModel):
  summary: str
  word_count: int
  key_points: str

def extract_text(url: str) -> str:
  """Extract text from a URL."""
  return f'Extracted content from {url}: ...'

summarizer = Agent(
    name='summarizer',
    mode='single_turn',
    output_schema=SummaryOutput,
    instruction=(
        'Summarize the document:\n'
        '1. Use extract_text to get content.\n'
        '2. Call finish_task with summary, word_count, key_points.\n'
        'Complete autonomously without user interaction.'
    ),
    description='Summarizes documents autonomously.',
    tools=[extract_text],
)

root_agent = Agent(
    name='coordinator',
    model='gemini-2.5-flash',
    sub_agents=[summarizer],
    instruction='Delegate summarization to summarizer via request_task_summarizer.',
)

Input and Output Schemas

Custom Schemas (Pydantic Models)

Define input_schema and/or output_schema with Pydantic BaseModel:

class TaskInput(BaseModel):
  query: str
  max_results: int = 10
  format: str = 'text'

class TaskOutput(BaseModel):
  results: str
  count: int
  status: str

agent = Agent(
    name='worker',
    mode='task',
    input_schema=TaskInput,    # Validates request_task_worker args
    output_schema=TaskOutput,  # Validates finish_task args
    ...
)

Default Schemas

When no custom schema is provided:

Default input (used by request_task_{name}):

class _DefaultTaskInput(BaseModel):
  goal: str | None = None
  background: str | None = None

Default output (used by finish_task):

class _DefaultTaskOutput(BaseModel):
  result: str

Auto-Generated Tools

request_task_{agent_name}

Auto-generated on the coordinator for each mode='task' or mode='single_turn' sub-agent. The tool name is request_task_{agent.name}.

• Parameters come from input_schema (or default: goal, background)
• Description includes the agent's description field
• Validates input against the schema before delegating

finish_task

Auto-generated on the task agent itself. Called by the task agent when work is complete.

• Parameters come from output_schema (or default: result)
• Validates output against the schema before signaling completion
• Sets tool_context.actions.finish_task with a TaskResult

Mixed-Mode Patterns

Combine task and single-turn agents under one coordinator:

# Interactive: user can discuss options
flight_searcher = Agent(
    name='flight_searcher',
    mode='task',
    input_schema=FlightSearchInput,
    output_schema=FlightSearchOutput,
    instruction='Search flights, discuss with user, then finish_task.',
    description='Searches and books flights interactively.',
    tools=[search_flights, book_flight],
)

# Autonomous: no user interaction
weather_checker = Agent(
    name='weather_checker',
    mode='single_turn',
    output_schema=WeatherOutput,
    instruction='Check weather and call finish_task. No user interaction.',
    description='Checks weather for a destination.',
    tools=[get_weather],
)

# Autonomous: no user interaction
hotel_finder = Agent(
    name='hotel_finder',
    mode='single_turn',
    output_schema=HotelOutput,
    instruction='Find hotels and call finish_task. No user interaction.',
    description='Finds hotels for a destination.',
    tools=[find_hotels],
)

root_agent = Agent(
    name='travel_planner',
    model='gemini-2.5-flash',
    sub_agents=[flight_searcher, weather_checker, hotel_finder],
    instruction=(
        'Help users plan trips:\n'
        '- request_task_weather_checker: autonomous weather check\n'
        '- request_task_hotel_finder: autonomous hotel search\n'
        '- request_task_flight_searcher: interactive flight booking'
    ),
)

Key Rules

• Both task sub-agents and coordinators use Agent from google.adk
• Each sub-agent needs a description (used in the auto-generated tool description)
• input_schema and output_schema are optional; defaults are provided
• Sub-agents inherit model from the coordinator if not set
• finish_task instructions are auto-injected into the task agent's LLM context
• Single-turn agents receive an extra instruction telling them no user replies will come

Task Mode vs Chat Mode

Source File Locations

Testing Workflow Agents Reference

Write unit tests for workflow agents using pytest with async support and the public InMemoryRunner from google.adk.runners.

Setup

# Install ADK + pytest + pytest-asyncio
pip install "google-adk>=2.0" pytest pytest-asyncio

# Or with uv
uv add "google-adk>=2.0" pytest pytest-asyncio

pyproject.toml:

[tool.pytest.ini_options]
asyncio_mode = "auto"

asyncio_mode = "auto" removes the need to mark every test with @pytest.mark.asyncio; if you'd rather mark each test explicitly, omit it.

Imports

All imports below are from the published google-adk package — no test-internal helpers required.

import pytest
from google.genai import types
from google.adk import Workflow
from google.adk.agents import LlmAgent
from google.adk.apps import App
from google.adk.apps.app import ResumabilityConfig
from google.adk.events import Event, RequestInput
from google.adk.runners import InMemoryRunner

A small run helper

Tests are tidier with a helper that drives one turn and collects events:

async def run(agent, text="hi", app_name="test_app"):
    runner = InMemoryRunner(agent=agent, app_name=app_name)
    session = await runner.session_service.create_session(
        app_name=app_name, user_id="u1"
    )
    msg = types.Content(role="user", parts=[types.Part(text=text)])
    events = []
    async for event in runner.run_async(
        user_id="u1", session_id=session.id, new_message=msg,
    ):
        events.append(event)
    return runner, session, events


def node_name(event):
    """Extract the node name from event.node_info.path.

    e.g. 'workflow@1/step@1' -> 'step'.
    """
    if not event.node_info:
        return None
    return event.node_info.path.split("/")[-1].split("@")[0]

In ADK 2.x, event.author is the enclosing workflow's name; the per-node identifier lives in event.node_info.path. Use node_name(event) to filter by the node that emitted an event.

Basic Workflow Test

async def test_simple_workflow():
    def step_one(node_input: str) -> str:
        return "step 1 done"

    def step_two(node_input: str) -> str:
        return "step 2 done"

    agent = Workflow(
        name="test_workflow",
        edges=[
            ("START", step_one),
            (step_one, step_two),
        ],
    )

    _, _, events = await run(agent)
    final = [e for e in events if node_name(e) == "step_two" and e.output][-1]
    assert final.output == "step 2 done"

Testing Conditional Routing

async def test_routing():
    def router(node_input: str):
        if "error" in node_input:
            return Event(output=node_input, route="error")
        return Event(output=node_input, route="success")

    def success_handler(node_input: str) -> str:
        return f"OK: {node_input}"

    def error_handler(node_input: str) -> str:
        return f"ERR: {node_input}"

    agent = Workflow(
        name="routing_test",
        edges=[
            ("START", router),
            (router, {"success": success_handler, "error": error_handler}),
        ],
    )

    _, _, evs_ok = await run(agent, text="all good")
    assert any(node_name(e) == "success_handler" for e in evs_ok)

    _, _, evs_err = await run(agent, text="error case")
    assert any(node_name(e) == "error_handler" for e in evs_err)

Testing HITL (Pause and Resume)

async def test_hitl_workflow():
    async def ask_user(ctx, node_input: str):
        yield RequestInput(message="Approve?", interrupt_id="ask")

    def after_approval(node_input) -> str:
        return f"Approved: {node_input}"

    agent = Workflow(
        name="hitl_test",
        edges=[
            ("START", ask_user),
            (ask_user, after_approval),
        ],
    )

    app = App(
        name="hitl_test_app",
        root_agent=agent,
        resumability_config=ResumabilityConfig(is_resumable=True),
    )
    runner = InMemoryRunner(app=app)
    session = await runner.session_service.create_session(
        app_name="hitl_test_app", user_id="u1"
    )

    # First turn: should pause with a RequestInput function call
    msg = types.Content(role="user", parts=[types.Part(text="start")])
    pause_events = []
    async for event in runner.run_async(
        user_id="u1", session_id=session.id, new_message=msg,
    ):
        pause_events.append(event)

    fc_events = [e for e in pause_events if e.get_function_calls()]
    assert fc_events, "expected an interrupt function call"
    fc = fc_events[-1].get_function_calls()[0]

    # Resume by responding to the function call
    response = types.Content(
        role="user",
        parts=[types.Part(function_response=types.FunctionResponse(
            id=fc.id, name=fc.name, response={"result": "yes"},
        ))],
    )
    resumed = []
    async for event in runner.run_async(
        user_id="u1", session_id=session.id, new_message=response,
    ):
        resumed.append(event)

    final = [e for e in resumed if node_name(e) == "after_approval"][-1]
    assert final.output == "Approved: yes"

Testing State Updates

Prefer asserting on the post-run session's state rather than reading state mid-flight:

async def test_state_management():
    def writer(node_input: str):
        return Event(output=node_input, state={"counter": 1})

    def reader(ctx, node_input):
        return f"counter={ctx.state['counter']}"

    agent = Workflow(
        name="state_test",
        edges=[("START", writer, reader)],
    )

    runner, session, events = await run(agent)
    final = [e for e in events if node_name(e) == "reader" and e.output][-1]
    assert final.output == "counter=1"

    # Or read state directly off the session after the run
    final_session = await runner.session_service.get_session(
        app_name="test_app", user_id="u1", session_id=session.id
    )
    assert final_session.state["counter"] == 1

Testing Parallel Execution

from google.adk.workflow import node

async def test_parallel_worker():
    def produce(node_input: str) -> list:
        return [1, 2, 3]

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

    def collect(node_input: list) -> str:
        return f"results: {node_input}"

    agent = Workflow(
        name="parallel_test",
        edges=[("START", produce, double, collect)],
    )

    _, _, events = await run(agent)
    final = [e for e in events if node_name(e) == "collect" and e.output][-1]
    assert final.output == "results: [2, 4, 6]"

Mocking LLM Agents

For unit tests that don't hit the real API, pass a fake BaseLlm to the LlmAgent constructor. The framework only requires the abstract generate_content_async method.

from google.adk.models.base_llm import BaseLlm
from google.adk.models.llm_response import LlmResponse
from google.genai import types


class FakeLlm(BaseLlm):
    def __init__(self, *, responses: list[str]):
        super().__init__(model="fake")
        self._responses = list(responses)

    async def generate_content_async(self, llm_request, stream=False):
        text = self._responses.pop(0)
        yield LlmResponse(content=types.Content(
            role="model", parts=[types.Part(text=text)],
        ))


async def test_llm_agent_with_fake():
    agent = LlmAgent(
        name="x",
        model=FakeLlm(responses=["ok"]),
        instruction="Help.",
    )
    _, _, events = await run(agent, text="hi")
    final = events[-1]
    assert final.content and final.content.parts[0].text == "ok"

If you only need to assert call shapes, monkeypatch the agent's canonical_model.generate_content_async with a mock instead.

Integration tests with a real model

Tag tests that hit a real model and skip them by default:

import os
import pytest

@pytest.fixture(scope="session", autouse=True)
def adk_env():
    if "GOOGLE_API_KEY" not in os.environ:
        pytest.skip("GOOGLE_API_KEY not set; skipping integration tests")
    os.environ.setdefault("GOOGLE_GENAI_USE_VERTEXAI", "FALSE")

@pytest.mark.integration
async def test_real_model():
    ...

Then pytest -m integration to run them, or pytest -m "not integration" to skip.

Testing Tips

• Create a fresh InMemoryRunner and session per test — runners hold state

and reuse causes cross-test interference.

• Use a unique app_name per test (e.g. request.node.name) to avoid

collisions across parallel pytest workers.

• Assert on event.node_info.path, not event.author. event.author is the

enclosing workflow's name; event.node_info.path identifies the exact node that emitted the event.

• Use event.is_final_response() to filter for "the agent's final message"

events.

• For workflows with a JoinNode, make sure every LLM agent feeding into it

has output_schema= set — otherwise the join buffer fails JSON serialization in tests that use DatabaseSessionService.

• Run with pytest -xvs while iterating (-x stop on first failure, -v

verbose, -s show prints) to debug event flow.