Google ADK 2.0 Tutorial: Graph Workflows and Multi-Agents

The first version of our procurement assistant let the model decide who approved a purchase. It read the request, guessed the amount band and transferred to a manager or director agent. It was right most of the time, which is the wrong standard for a rule the finance team had already written down as two numbers: under 5 million rupiah is automatic, under 50 million needs a manager, anything above needs a director.

This Google ADK 2.0 tutorial rebuilds that flow on the new graph engine, where the LLM only extracts the request and plain Python does the routing. I ran every snippet against google-adk 2.10.0 on Python 3.11, with the intake agent swapped for a function stub so the deterministic path could run without a model. Where the behaviour surprised me, the post says so; everything else traces to the official ADK 2.0 docs.

What changed in ADK 2.0: agents became nodes

ADK 2.0 replaced hierarchical agent execution with a graph engine. Agents, tools and plain functions are all nodes, and BaseAgent now extends BaseNode. Python reached GA on 19 May 2026, Go on 30 June 2026 and TypeScript on 21 August 2026, and Python has shipped a minor release roughly every two weeks since, reaching 2.10.0 in late September. The release gives you three ways to compose nodes.

The question that picks the row is simple: could a junior analyst follow the rule from a written policy? If yes, it belongs in a graph or a dynamic workflow, and the model should not be the one choosing the branch. The approval thresholds pass that test, so they go in code. Turning a vague message into a vendor, an amount and a cost centre does not, so that step stays with the model.

Google ADK 2.0 tutorial: a procurement approval graph

The whole policy fits in one Workflow. The edges list reads top to bottom: START feeds the intake agent, its structured output feeds the amount router, and each route value maps to a branch. Every approval branch converges on a second router that splits approved from rejected.

# agent.py — google-adk 2.10.0, Python 3.11
from pydantic import BaseModel, Field
from google.adk import Agent, Workflow, Event, Context
from google.adk.events import RequestInput
from google.adk.workflow import DEFAULT_ROUTE

class PurchaseRequest(BaseModel):
    vendor: str
    amount_idr: int = Field(ge=0)
    cost_center: str
    justification: str

class Decision(BaseModel):
    approved: bool
    approver: str
    note: str | None = None   # Optional, not str = "" — see below

# The one LLM node: free text in, a validated PurchaseRequest out.
intake_agent = Agent(
    name="intake_agent",
    model="gemini-flash-latest",
    instruction="Extract the purchase request from the user's message.",
    output_schema=PurchaseRequest,
)

def route_by_amount(node_input: PurchaseRequest, ctx: Context):
    ctx.state["request"] = node_input.model_dump()   # later nodes read it back
    if node_input.amount_idr < 5_000_000:
        return Event(route="AUTO", output=node_input)
    if node_input.amount_idr < 50_000_000:
        return Event(route="MANAGER", output=node_input)
    return Event(route="DIRECTOR", output=node_input)

def auto_approve(node_input: PurchaseRequest):
    return Event(output=Decision(approved=True, approver="policy:auto"))

def manager_approval(node_input: PurchaseRequest):
    # Pauses the run. The human's reply becomes the NEXT node's input.
    yield RequestInput(message="Manager approval",
                       payload=node_input.model_dump(),
                       response_schema=Decision)

def director_approval(node_input: PurchaseRequest):
    yield RequestInput(message="Director approval",
                       payload=node_input.model_dump(),
                       response_schema=Decision)

# ADK coerces node_input to the annotation, so the human's JSON reply
# arrives here as a validated Decision.
def decision_router(node_input: Decision):
    route = "APPROVED" if node_input.approved else "REJECTED"
    return Event(route=route, output=node_input)

def create_po(node_input: Decision, ctx: Context):
    return Event(message=f"PO drafted for {ctx.state['request']['vendor']}")

def notify_rejection(node_input: Decision):
    return Event(message=f"Rejected by {node_input.approver}")

def needs_review():
    return Event(message="Unrecognised route; sent to procurement inbox")

root_agent = Workflow(
    name="procurement_approval",
    edges=[
        ("START", intake_agent, route_by_amount),
        (route_by_amount, {
            "AUTO": auto_approve,
            "MANAGER": manager_approval,
            "DIRECTOR": director_approval,
            DEFAULT_ROUTE: needs_review,     # never ship a router without it
        }),
        (auto_approve, decision_router),
        (manager_approval, decision_router),
        (director_approval, decision_router),
        (decision_router, {"APPROVED": create_po, "REJECTED": notify_rejection}),
    ],
)

Three details carry the design. The intake agent uses output_schema, so the router receives a validated PurchaseRequest rather than prose. ADK coerces each node_input to the parameter's type annotation, which is why a node can return a plain dict and the next one still gets a model. And route_by_amount writes the request into ctx.state, so create_po can read the vendor two hops later without threading it through every return value.

Pausing for a human with RequestInput

A manager does not answer in milliseconds, so the approval node cannot block. Yielding RequestInput ends the current run and emits an event whose content is a function call named adk_request_input, carrying the message, your payload and an interrupt id that is also listed in long_running_tool_ids. To resume, send a function response with that id back into the same session.

import asyncio
from google.adk import Runner
from google.adk.sessions import InMemorySessionService
from google.genai import types
from agent import root_agent

async def main():
    sessions = InMemorySessionService()
    runner = Runner(app_name="proc", node=root_agent, session_service=sessions)
    session = await sessions.create_session(app_name="proc", user_id="budi")

    ask = types.Content(role="user", parts=[types.Part(
        text="Need 12 juta for printer paper from PT Kertas, cost centre OPS")])

    pending = None
    async for ev in runner.run_async(user_id="budi", session_id=session.id,
                                     new_message=ask):
        if ev.long_running_tool_ids:            # the RequestInput event
            pending = next(iter(ev.long_running_tool_ids))
            print("paused at", ev.node_info.path)
            # -> procurement_approval@1/manager_approval@1

    # Hours later, from the approval UI. Same session id, same interrupt id.
    answer = types.Content(role="user", parts=[types.Part(
        function_response=types.FunctionResponse(
            id=pending, name="adk_request_input",
            response={"approved": False, "approver": "siti"}))])

    async for ev in runner.run_async(user_id="budi", session_id=session.id,
                                     new_message=answer):
        if ev.content and ev.content.parts[0].text:
            print(ev.content.parts[0].text)     # -> Rejected by siti

asyncio.run(main())

• The reply is handed to the successor node, not to the node that paused, so manager_approval never runs a second time. That is the default for function nodes, where rerun_on_resume is false.
• response_schema is enforced on the way back in. A reply of approved set to maybe made run_async raise WorkflowDataError with the Pydantic errors, so wrap the resume call and send the error back to your approval UI.
• Defaults do not survive the round trip. My first Decision model had note as str with an empty-string default; the resumed value carried note as None, and the annotated decision_router failed with a ValidationError event. Declaring note as optional with a None default fixed it.

The route nobody mapped ends the run silently

This is the failure I would most like someone to have warned me about. When a router returns a route value that is not a key in its map, ADK does not raise and does not emit an error event. The router's output event appears, and then the run just ends. In a test I returned an unmapped value and got exactly one event back.

# Wrong: a router whose value is not a key. No exception, no error event —
# the run emits router's output and simply stops.
(router, {"AUTO": auto_approve, "MANAGER": manager_approval})
# router returns Event(route="Manager")   <- casing drift from an LLM, or a new tier

# Right: catch everything you did not name.
from google.adk.workflow import DEFAULT_ROUTE
(router, {"AUTO": auto_approve, "MANAGER": manager_approval,
          DEFAULT_ROUTE: needs_review})

Sessions that survive a restart

Pausing is only useful if the pause outlives the process. InMemorySessionService loses everything on restart, so production needs DatabaseSessionService. I created a session, ran it to the manager interrupt, then built a brand-new service and Runner on the same SQLite file and sent the approval. The workflow resumed at record_decision with the request still in state.

# pip install "google-adk[db]==2.10.0" aiosqlite greenlet
#                                               ^^^^^^^^ not pulled in by [db];
# without it you get: ValueError: Failed to create database engine for URL ...
from google.adk.sessions import DatabaseSessionService

DB_URL = "sqlite+aiosqlite:///./proc.db"   # the service needs an async driver, not plain sqlite://

# Process 1: run until the manager_approval interrupt, then exit.
sessions = DatabaseSessionService(db_url=DB_URL)
runner = Runner(app_name="proc", node=root_agent, session_service=sessions)

# Process 2, after a restart: a brand-new service and runner on the same file.
sessions = DatabaseSessionService(db_url=DB_URL)
runner = Runner(app_name="proc", node=root_agent, session_service=sessions)
# run_async(..., new_message=<function_response for the stored interrupt id>)
# -> record_decision runs; nothing before the interrupt is executed again.

# What lands on disk (tables: sessions, events, app_states, user_states,
# adk_internal_metadata). Each events.event_data row is JSON like:
# {"author": ..., "node_info": {"path": "procurement_approval@1/route_by_amount@1",
#   "output_for": [...]}, "output": {...}, "actions": {...}}

Getting there cost me one misleading error. The db extra installs SQLAlchemy, which resolved to 2.1.1, and the asyncio layer of that release needs greenlet, which nothing pulled in. ADK reported only that it failed to create the database engine; the real cause was an ImportError underneath. Installing greenlet fixed it. The event rows also show why custom session stores break on 2.0: node_info and output now sit inside every stored event, and a validator that rejects unknown properties will reject them.

When the graph is wrong: dynamic and collaborative workflows

A real approval matrix is rarely three fixed boxes. Above 500 million rupiah our chain adds a CFO, and drawing a graph for every chain length gets ugly quickly. A dynamic workflow expresses it as a loop, with each approval delegated to a child node through ctx.run_node.

# Same module as before: reuses PurchaseRequest and intake_agent
from google.adk import Context, Workflow
from google.adk.events import RequestInput
from google.adk.workflow import node

def approval_chain(amount_idr: int) -> list[str]:
    chain = ["manager"]
    if amount_idr >= 50_000_000:
        chain.append("director")
    if amount_idr >= 500_000_000:
        chain.append("cfo")
    return chain

@node(name="ask_approver")
def ask_approver(node_input: dict):
    yield RequestInput(message=f"{node_input['role']} approval", payload=node_input)

@node(rerun_on_resume=True)          # required for an orchestrator using run_node
async def approval_loop(ctx: Context, node_input: PurchaseRequest):
    for role in approval_chain(node_input.amount_idr):
        # Each resume re-enters this loop from the top. Completed ask_approver
        # calls return their stored answer instantly instead of pausing again.
        decision = await ctx.run_node(
            ask_approver, node_input={"role": role, **node_input.model_dump()})
        if not decision["approved"]:
            return f"rejected by {role}"
    return "approved"

root_agent = Workflow(name="approval_chain",
                      edges=[("START", intake_agent, approval_loop)])

# Trace for 600,000,000 IDR, three resumes:
#   body: manager                      -> pause
#   body: manager, director            -> pause
#   body: manager, director, cfo       -> pause
#   body: manager, director, cfo       -> "approved"
# Anything with side effects belongs in a child node, never in this body.

The trace in the comment is from a real run and it is the thing to understand: on every resume the orchestrator body starts again from the top, and finished children return their stored answers instead of pausing again. The checkpointing lives in the children. Anything with a side effect, such as an email, an ERP write or a counter, must be its own node, or it will fire once per resume. When the work itself is open-ended, such as finding vendors and collecting quotes, a coordinator with subagents fits better.

vendor_lookup = Agent(
    name="vendor_lookup",
    mode="single_turn",                 # no user interaction, can run in parallel
    tools=[search_vendor_master],
)

quote_collector = Agent(
    name="quote_collector",
    mode="task",                        # may ask clarifying questions, auto-returns
    input_schema=QuoteRequest,
    output_schema=QuoteSummary,
    tools=[request_quote, read_quote_inbox],
)

root_agent = Agent(                     # no mode on the root
    name="sourcing_coordinator",
    sub_agents=[vendor_lookup, quote_collector],
)

• chat is the default mode: the subagent talks with the user and must hand control back explicitly.
• task allows clarifying questions and returns automatically when finished. It cannot have its own subagents and is disabled inside graph workflows in Python 2.0.
• single_turn allows no user interaction, returns immediately and is the only mode that can run in parallel.

The coordinator gets one generated tool per subagent, named after it, and task and single_turn agents run in isolated session branches, so parallel siblings cannot see each other's work. Do not set a mode on the root agent. The model chooses which subagent to call, which is exactly why approvals stay in the graph and sourcing goes here.

Testing the graph and evaluating the agent

Split the testing along the same line as the design. Deterministic nodes are ordinary Python functions, so test them with pytest and no API key. The one model-backed node is where adk eval earns its place: tool_trajectory_avg_score compares tool calls against an expected list, and response_match_score compares the final answer.

// tests/test_config.json — for the intake_agent eval set
{
  "criteria": {
    "tool_trajectory_avg_score": { "threshold": 1.0, "match_type": "IN_ORDER" },
    "response_match_score": 0.8
  }
}

# Run it
adk eval procurement/ tests/intake.evalset.json \
  --config_file_path=tests/test_config.json --print_detailed_results

# The routing nodes need no model at all: plain pytest
def test_director_threshold():
    assert approval_chain(50_000_000) == ["manager", "director"]
    assert approval_chain(49_999_999) == ["manager"]

The trajectory criterion defaults to a threshold of 1.0 with an exact match, meaning no extra or missing calls. IN_ORDER accepts extra calls between the expected ones, and ANY_ORDER ignores sequence, so pick the loosest mode that still catches the regression you care about. response_match_score defaults to 0.8. For the intake agent I care that it never skips the vendor lookup, so IN_ORDER is the honest setting.

Migrating from ADK 1.x without breaking sessions

Most 1.x agents keep working, but four changes bite in practice.

1. Update custom session stores and strict JSON validators for the new Event fields: node_info, output, route and isolation scope in every language, plus RequestedInput in Go.
2. In Python, overrides of _run_async_impl are bypassed. Move that logic into before and after agent callbacks, yield events instead of appending to context.session.events, and stop catching BaseException so retries can work.
3. In Go, change the module path to google.golang.org/adk/v2 and pass the context as the first argument to session.NewEvent.
4. In TypeScript, SequentialAgent, ParallelAgent and LoopAgent are deprecated in favour of graph workflows, and InvocationContext.agent may now be undefined.

// Go 1.x
import "google.golang.org/adk/session"
ev := session.NewEvent(ctx.InvocationID())

// Go 2.0 — new module path, and NewEvent takes the context first
import "google.golang.org/adk/v2/session"
ev := session.NewEvent(ctx, ctx.InvocationID())

# Python: pin 1.x while you migrate
pip install "google-adk~=1.0"
# TypeScript
npm install @google/adk@^1.6.0

The rule I took from the rebuild: let the model read, let the code decide. In ADK 2.0 that is a Workflow whose only LLM node turns text into a schema, with every threshold in a route map that has a DEFAULT_ROUTE. Persist sessions in a database before the first human pause, and keep side effects in child nodes so a resume cannot repeat them.