google--adk-python/.agents/skills/adk-agent-builder/references/dynamic-nodes.md

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.