Dynamic Node Scheduling

Dynamic node scheduling allows you to execute workflow nodes dynamically at runtime using ctx.run_node(). This enables imperative workflow construction using standard Python control flow instead of static graph edges.

Introduction

While static graph definitions (Workflow(edges=[...])) are suitable for many structured tasks, some scenarios require more flexibility. For example, you might need to:

• Loop a set of nodes until a condition is met (e.g., generator-evaluator loops).
• Run a variable number of tasks in parallel based on runtime input (dynamic fan-out).
• Conditionally execute nodes based on complex logic that is difficult to express in static edges.

ctx.run_node() allows a parent node to execute a child node (which can be a function, an Agent, or another Workflow) and await its result.

Get started

The following example demonstrates how to dynamically execute a child agent from a parent node.

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

# Define a child agent
generate_headline = Agent(
    name="generate_headline",
    instruction="Write a catchy headline about the topic in the user message.",
)


# Define the parent orchestrator node (MUST have rerun_on_resume=True)
@node(rerun_on_resume=True)
async def orchestrate(ctx: Context, node_input: str) -> str:
  # Dynamically execute the child agent and await its output
  headline = await ctx.run_node(generate_headline, node_input=node_input)

  yield Event(output=headline)

# Build the workflow
root_agent = Workflow(
    name="root_agent",
    edges=[("START", orchestrate)],
)

How it works

When await ctx.run_node(node_like, ...) is called:

1. Orchestrator Registration: The workflow's DynamicNodeScheduler registers the child node execution.
2. State Tracking: The execution state and events of the child node are tracked under the parent node's path (e.g., parent_node@1/child_node@1).
3. Resumption Support: If the child node interrupts (e.g., waiting for user input), the parent node is also paused. When the workflow resumes, the parent node is re-run from the beginning (rerun_on_resume=True), but previous successful ctx.run_node() calls are replayed from history (cached outputs are returned) to avoid re-executing completed steps.

Input Mapping

The node_input passed to ctx.run_node(node, node_input=value) is delivered differently depending on the type of the child node:

• Python Functions / FunctionNodes: The value is passed directly to the function parameter named node_input. Other parameters are bound from the session state (default mode).
• Agents (Single-Turn Mode): The value is converted to a user-role message (types.Content) and appended to the session events history. The agent receives it as the incoming user message.
• Agents (Task Mode): The value is set as user_content in the InvocationContext, serving as the fallback first user turn for the task agent if it wasn't triggered by a tool call.

Requirements & Rules

1. rerun_on_resume=True is Mandatory for Parents

Any node that calls ctx.run_node() must be configured with rerun_on_resume=True. If the parent node does not have this setting, calling ctx.run_node() will raise a ValueError at runtime.

2. Function Parameter Mapping (node_input vs. Dict Binding)

By default, functions wrapped as nodes look up their arguments in the session state (state binding). However, the node_input argument passed to ctx.run_node(..., node_input=value) is passed directly to the node.

How you receive this input depends on how you define your function:

Pass-through node_input (Default)

To receive the raw value directly, the function's parameter must be named exactly node_input.

# Correct: receives the raw value passed to node_input
def my_worker(node_input: str):
  return f"Done: {node_input}"

# Incorrect: will fail because it tries to look up 'data' in session state
def my_worker(data: str):
  return f"Done: {data}"

Binding Dictionary Keys to Parameters (parameter_binding='node_input')

If you pass a dictionary to node_input (e.g., node_input={'foo': 'bar'}) and want to bind its keys to individual function parameters (e.g., def my_worker(foo: str)), you must configure the node with parameter_binding='node_input'.

You can configure this using the @node decorator with parameter_binding='node_input':

from google.adk.workflow import node

# Decorate with parameter_binding='node_input'
@node(parameter_binding='node_input')
def my_worker(foo: str):
  return f"Done: {foo}"

# Call via ctx.run_node
result = await ctx.run_node(my_worker, node_input={'foo': 'bar'}) # foo gets 'bar'

3. Nested Dynamic Nodes

If a dynamically scheduled node itself calls ctx.run_node(), it becomes a parent and must also have rerun_on_resume=True. You should decorate the nested function with @node(rerun_on_resume=True) to ensure it has this property when executed:

from google.adk.workflow import node

@node(rerun_on_resume=True)
async def inner_parent(ctx: Context):
  # Calls another dynamic node internally
  result = await ctx.run_node(some_child)
  yield Event(output=result)

# In the outer parent:
await ctx.run_node(inner_parent)

4. Generator Returns

In nodes that use yield (generators), you cannot use return value to produce the final output of the node due to Python syntax constraints. You must yield Event(output=value) instead.

Method Signature

async def run_node(
    self,
    node: NodeLike,
    node_input: Any = None,
    *,
    use_as_output: bool = False,
    run_id: str | None = None,
    use_sub_branch: bool = False,
    override_branch: str | None = None,
) -> Any:

Parameters

Advanced Applications

Dynamic Fan-Out (Parallel Execution)

You can perform dynamic fan-out by scheduling multiple tasks in parallel using asyncio.gather. When doing this, you must set use_sub_branch=True to isolate the events of each parallel execution.

import asyncio
from google.adk import Context, Event, Agent
from google.adk.workflow import node

worker = Agent(name="worker", instruction="Process {node_input}")

@node(rerun_on_resume=True)
async def parallel_orchestrator(ctx: Context, node_input: list[str]):
  tasks = []
  for topic in node_input:
    tasks.append(
        ctx.run_node(
            worker,
            node_input=topic,
            use_sub_branch=True, # Critical for parallel isolation
        )
    )

  # Await all tasks concurrently
  results = await asyncio.gather(*tasks)
  yield Event(output=results)

Best Practices

• Avoid Unsupervised Tasks: Always await ctx.run_node() directly (or via asyncio.gather). Do not wrap it in asyncio.create_task() without awaiting it, as errors will be swallowed, and tasks won't be cancelled if the workflow is interrupted.
• Manage Side Effects and Resumption: Because a parent node with rerun_on_resume=True is executed from the beginning on resumption, any code with side effects (e.g., database writes, API calls) in the parent node will run again.
  • Best Practice: Keep the parent orchestrator node's logic as light as possible, containing mostly control flow and ctx.run_node calls.
  • Best Practice: Move any logic with side effects into dedicated child nodes and execute them via ctx.run_node. Since completed child nodes are cached and replayed, their side effects will not be executed again on resumption.

Limitations

• Replay Overhead: Because the parent node is re-run from the beginning on resume, long-running parent node logic (outside of ctx.run_node calls) will be re-executed. Keep the orchestrator node logic light and delegate heavy lifting to child nodes.

Related samples

• Dynamic Nodes Sample
• Dynamic Fan-Out / Fan-In Sample