dynamic-workflow-builder

dynamic-workflow-builder

Code-defined workflow orchestration introduced in ADK 2.0. The graph topology itself is computed at runtime by your Python code — not declared up-front like GraphAgent.

When to use

• Number/order of steps depends on data not known at design time
• You want to write idiomatic Python control flow (if/while/for)
• Mixing ADK agents with arbitrary Python (DB queries, API calls)
• Custom retry / backoff / circuit-breaker logic per agent

When NOT to use

• Topology is known and static → graph-workflow-designer
• Simple linear chain → SequentialAgent
• Fixed iteration → LoopAgent

Template

from google.adk.agents import BaseAgent, LlmAgent
from google.adk.agents.invocation_context import InvocationContext
from google.adk.events import Event
from typing import AsyncIterator


class DynamicResearchAgent(BaseAgent):
    """Iteratively expands research until coverage threshold is met."""

    name: str = "dynamic_researcher"

    def __init__(self):
        super().__init__()
        self.searcher = LlmAgent(
            name="searcher",
            model="gemini-2.5-flash",
            instruction="Search for one specific subtopic and write findings.",
        )
        self.evaluator = LlmAgent(
            name="evaluator",
            model="gemini-2.5-flash",
            instruction=(
                "Score coverage 0-10. If <8, write missing subtopic to "
                "state['next_query']. If >=8, set state['done']=True."
            ),
        )

    async def _run_async_impl(
        self, ctx: InvocationContext
    ) -> AsyncIterator[Event]:
        attempts = 0
        max_attempts = 6
        while attempts < max_attempts:
            attempts += 1
            async for ev in self.searcher.run_async(ctx):
                yield ev
            async for ev in self.evaluator.run_async(ctx):
                yield ev
            if ctx.session.state.get("done"):
                yield Event(author=self.name, content=f"Converged after {attempts} rounds.")
                return
        yield Event(author=self.name, content=f"Stopped at max_attempts={max_attempts}.")

Wire as root

root_agent = DynamicResearchAgent()

Mixing in non-ADK Python

async def _run_async_impl(self, ctx):
    # Custom Python: hit your own DB
    rows = await my_db.query(ctx.session.state["search_term"])
    ctx.session.state["rows"] = rows

    # Then run an ADK agent
    async for ev in self.summarizer.run_async(ctx):
        yield ev

Validation

• Always cap iterations (max_attempts) — protect against runaway loops
• Yield events on entry/exit so the run UI shows progress
• Read/write ctx.session.state for cross-step communication
• Unit-test the orchestration logic with a mocked InvocationContext