Build dynamic workflows in Python

When you need this. A documentation change may touch one page or many. Each failing page may need a different number of fix attempts. We will use Python to start one child workflow per page and stop each fix loop when its checks pass or its limit is reached.

A dynamic workflow chooses child executions while it runs. ADK’s @node marks a function as a node, and ctx.run_node executes a child with a recorded input and output. We can use familiar while loops and asyncio.gather around those calls.

Our example checks Markdown for a missing front-matter title, HTTP links and skipped heading levels. Only a page with findings reaches the fixer agent. The result is proposed Markdown; the original files stay unchanged.

Prepare the page-review project

You need Python 3.12 or later and a Gemini API key. Use Download code, extract the ZIP, and open its adk-dynamic-workflow-lesson directory. Keep the Python files and fixtures together. All commands in this lesson run from that directory.

python3 -m venv .venv
source .venv/bin/activate
python -m pip install -r requirements.txt

On Windows PowerShell, activate with .venv\Scripts\Activate.ps1.

The Python client reads exported environment variables. The bundled .env.example documents them; it is not loaded by the client. Set GOOGLE_API_KEY privately, then select the Gemini API in this terminal:

unset GOOGLE_GENAI_USE_VERTEXAI DOCS_MODEL
export GOOGLE_GENAI_USE_ENTERPRISE=0

Sending a request to the model uses your provider account. The offline checks later in the lesson need no key.

The bundled docs/ folder contains the pages to review. Here is the install page before a fix:

---
section: setup
---
# Install

## Requirements

Python 3.12 or later.

#### Optional: a local model

Set `OLLAMA_API_BASE` to use a local model instead of Gemini.

Write the dynamic workflow

The workflow separates the deterministic lint check from the model-based rewrite. Read lint_page first, then follow its calls inside review_page.

"""Lint the docs pages a change touched, in parallel, and let Gemini fix what fails."""
import asyncio
import re
from pathlib import Path

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

DOCS = Path(__file__).with_name("docs")
MAX_ROUNDS = 2
MAX_CONCURRENT_PAGES = 4  # review_page children in flight at once


class Page(BaseModel):
    path: str
    content: str


class FixRequest(Page):
    findings: list[str]


class FixedPage(BaseModel):
    content: str


@node
def lint_page(node_input: Page):
    """Deterministic checks; the list of findings is the node's output."""
    findings = []
    front = re.match(r"^---\n(.*?)\n---\n", node_input.content, re.S)
    if not front or not re.search(r"^title: \S", front[1], re.M):
        findings.append("front matter has no title")
    findings += [f"insecure link {url}"
                 for url in re.findall(r"\]\((http://[^)]+)\)", node_input.content)]
    levels = [len(marks) for marks in re.findall(r"^(#+) ", node_input.content, re.M)]
    findings += [f"heading jumps from h{a} to h{b}"
                 for a, b in zip(levels, levels[1:]) if b > a + 1]
    return findings


def make_fixer(model=None, retry_delay=2.0, timeout=60.0):
    agent = Agent(
        name="fix_page",
        model=model or "gemini-3.8-flash",  # or LiteLlm(model="ollama_chat/gemma4:e2b")
        generate_content_config=types.GenerateContentConfig(max_output_tokens=1024, thinking_config=types.ThinkingConfig(thinking_level="low")),
        input_schema=FixRequest,
        output_schema=FixedPage,
        instruction=("Rewrite the Markdown page so that every finding is fixed. "
                     "Change nothing else, and return the whole page."),
    )
    # A model call can fail with a 429 or 503, or hang: retry it, and bound each attempt.
    return node(agent, retry_config=RetryConfig(max_attempts=3, initial_delay=retry_delay),
                timeout=timeout)


def build_workflow(model=None, max_concurrent=MAX_CONCURRENT_PAGES, **fixer_options) -> Workflow:
    fixer = make_fixer(model, **fixer_options)

    @node(rerun_on_resume=True)
    async def review_page(ctx: Context, node_input: str):
        """Lint one page; while it fails, ask the fixer and lint again, at most MAX_ROUNDS times."""
        page = Page(path=node_input, content=(DOCS / node_input).read_text())
        findings = await ctx.run_node(lint_page, page)
        found, rounds = list(findings), 0
        while findings and rounds < MAX_ROUNDS:
            rounds += 1
            fixed = await ctx.run_node(fixer, FixRequest(**page.model_dump(), findings=findings))
            page = Page(path=page.path, content=fixed["content"])
            findings = await ctx.run_node(lint_page, page)
        return {"path": page.path, "found": found, "rounds": rounds,
                "left": findings, "content": page.content}

    @node(rerun_on_resume=True)
    async def lint_changed_pages(ctx: Context, node_input: str):
        """Fan out one review_page child per changed Markdown file, at most max_concurrent at once."""
        paths = [path for path in node_input.split() if path.endswith(".md")]
        limit = asyncio.Semaphore(max_concurrent)

        async def review(path):
            async with limit:  # a child starts only when a slot is free
                return await ctx.run_node(review_page, path)

        results = await asyncio.gather(*(review(path) for path in paths))
        lines = [f"{r['path']}: " + ("clean" if not r["found"] else
                 f"fixed in {r['rounds']} round(s): {r['found']}" if not r["left"] else
                 f"still failing after {r['rounds']} rounds: {r['left']}") for r in results]
        yield Event(message="\n".join(lines))

    return Workflow(name="docs_lint", edges=[("START", lint_changed_pages)])

lint_page returns a list of findings. review_page calls the fixer only while that list is nonempty and rounds < MAX_ROUNDS. It checks the returned Markdown again after every fix. The model’s claim that a page is fixed has no role in the exit condition.

The parent starts a review_page child for each path with asyncio.gather. An asyncio.Semaphore allows at most MAX_CONCURRENT_PAGES children to run at once. That limit applies to active page reviews; each review can still use multiple model attempts.

Both parent functions use rerun_on_resume=True. If a child pauses, ADK reruns its parent on resume and returns saved outputs for completed child executions. Keep the sequence of child calls stable so the resumed execution can find them.

The fixer node has RetryConfig(max_attempts=3) and a timeout per attempt. These limits solve different problems: a failed attempt can retry; a valid response that leaves lint findings can trigger another fix round. The current two-round loop is separate from the retry count.

A transient model error. The test model raises ConnectionError twice, and the third attempt succeeds.

Retries exhausted. After the configured attempts fail, the last ConnectionError reaches run_async and the run stops.

The model never fixes the page. The loop stops at MAX_ROUNDS and reports the remaining findings instead of looping forever.

Too many pages at once. The semaphore keeps waiting children out of the active review section. The test sets max_concurrent=1 and checks that model calls no longer overlap.

An except clause hides the error. If the node catches an exception and returns a fallback, the retry wrapper sees a completed attempt and does not retry it. Catch only failures you can actually resolve.

Review the changed pages

python lint_client.py quickstart.md install.md cli.md

For each review_page output, inspect found, rounds, left and content. A clean page should have no initial findings and use no fix round. A fixed page should have an empty left list. If findings remain, the loop reached its bound.

Compare the proposed content with the original file before accepting it. These lint rules check document structure and link syntax, not factual accuracy or whether an HTTPS URL works.

To focus on one page, use python lint_client.py install.md. The client accepts only existing Markdown pages inside the bundled docs/ directory and rejects names containing whitespace before starting a Runner.

"""Choose bundled docs pages and inspect proposed changes without rewriting the files."""
import argparse
import asyncio
import json
import os
from google.adk.runners import InMemoryRunner
from lint_changed import DOCS, build_workflow
from harness import run

def parse_options(argv=None):
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("pages", nargs="*", default=["quickstart.md", "install.md", "cli.md"])
    args = parser.parse_args(argv)
    if not args.pages: parser.error("choose at least one page")
    for name in args.pages:
        page = (DOCS / name).resolve()
        if any(char.isspace() for char in name) or not name.endswith(".md") or not page.is_relative_to(DOCS.resolve()) or not page.is_file():
            parser.error("choose an existing Markdown page inside this lab's docs folder")
    return args

async def run_one(options, model=None):
    workflow = build_workflow(model=model or os.environ.get("DOCS_MODEL") or None)
    runner = InMemoryRunner(agent=workflow, app_name="docs_assistant")
    try:
        events, session_id = await run(workflow, " ".join(options.pages), runner=runner)
        result = {"session_id": session_id, "pages": options.pages, "events": events}
        print(json.dumps(result, ensure_ascii=False, indent=2))
        return result
    finally:
        await runner.close()

if __name__ == "__main__": asyncio.run(run_one(parse_options()))

The client prints events and closes its InMemoryRunner. It never writes the returned Markdown to the source files.

Test retries, timeouts and the loop bound

python -m unittest -v test_lint_changed.py test_client.py

Saved replies test the expected lint/fix/lint flow. Failing, delayed and unchanged replies exercise the limits without contacting a provider. Parallel requests are matched by their input rather than by arrival order.

A model call hangs. Each attempt times out with NodeTimeoutError. A timeout is retried under this policy, so three timed-out attempts can take three timeout periods plus backoff delays.

RetryConfig retries exceptions by default. Use its exceptions option to narrow that policy for a real application. If you also configure retries in the model client, account for both layers when deciding how many requests a failure can cause.

Try replacing a fake fix with the unchanged input in a separate test copy. Predict the final rounds and left, then run the matching offline check. The bound should stop the loop while preserving the unresolved findings.

Run Python nodes and bounded child workflows

A child execution sits under its parent’s node_info.path. This excerpt from the sample fixture shows the nested lint and fixer calls:

[
  {
    "author": "docs_lint",
    "path": "docs_lint@1/lint_changed_pages@1/review_page@1/lint_page@1"
  },
  {
    "author": "docs_lint",
    "path": "docs_lint@1/lint_changed_pages@1/review_page@2/lint_page@1"
  },
  {
    "author": "docs_lint",
    "path": "docs_lint@1/lint_changed_pages@1/review_page@3/lint_page@1"
  },
  {
    "author": "docs_lint",
    "path": "docs_lint@1/lint_changed_pages@1/review_page@3"
  },
  {
    "author": "fix_page",
    "path": "docs_lint@1/lint_changed_pages@1/review_page@1/fix_page@1"
  },
  {
    "author": "docs_lint",
    "path": "docs_lint@1/lint_changed_pages@1/review_page@1/lint_page@2"
  },
  {
    "author": "docs_lint",
    "path": "docs_lint@1/lint_changed_pages@1/review_page@1"
  },
  {
    "author": "fix_page",
    "path": "docs_lint@1/lint_changed_pages@1/review_page@2/fix_page@1"
  },
  {
    "author": "docs_lint",
    "path": "docs_lint@1/lint_changed_pages@1/review_page@2/lint_page@2"
  },
  {
    "author": "docs_lint",
    "path": "docs_lint@1/lint_changed_pages@1/review_page@2"
  },
  {
    "author": "docs_lint",
    "path": "docs_lint@1/lint_changed_pages@1"
  }
]

ctx.run_node gives ADK a checkpoint boundary around the child. It does not turn ordinary Python statements into resumable steps. Keep side effects inside deliberate nodes and account for code that runs again when a parent resumes.

Use fixed graph edges when the steps are known before a run. Use dynamic nodes when the input decides the number of children or iterations. In both forms, write the exit condition and resource limits in application code.

Next, delegate to sub-agents when a model should choose a specialist instead of Python choosing a child.

References

The code uses the installed google-adk 2.12.0 node, Context and RetryConfig interfaces.

• ADK dynamic workflows: child nodes and resume behavior.
• ADK 2.0 overview: retries and exceptions.
• Python asyncio.gather: concurrent awaitables.
• Python asyncio.Semaphore: bounded access.