Google ADK 2.0: Graphs, Routing and Human Approval

ADK 2.0 changed how Google's Agent Development Kit runs agents. In ADK 1.x an agent was a standalone executor. In 2.0, agents, tools and plain functions are nodes in a workflow graph, and a scheduler runs them: it can branch, fan out, pause for a person and resume later. Graph workflows are supported in ADK Python, TypeScript and Go 2.0. This post walks through the patterns we use most, with Python examples based on the ADK 2.0 docs.

Three ways to compose agents

• Graph-based workflows: a declarative graph of nodes and edges with explicit routing. Best for structured processes such as ticket triage or claims handling.
• Dynamic workflows: orchestration in your own code, with loops, conditionals and recursion, for control flow that is too irregular for a static graph.
• Prebuilt workflow agents: SequentialAgent, ParallelAgent and LoopAgent, for common patterns without assembling a graph.

All three mix language-model agents with deterministic code. That matters in production: the reply can come from a model, but the refund limit check should be a function you can unit test.

How data moves through a graph

A Workflow takes a list of edges. Each tuple is a chain of nodes that run in order, starting from the special START node, which hands the user's message to the first node. A node passes data to the next one by returning a value or an Event(output=...). Event(message=...) is for the user and is not passed on. Event(route=...) chooses a branch, and route and output can travel in the same event.

from google.adk import Agent, Event, Workflow

summarise = Agent(
    name="summarise_ticket",
    model="gemini-flash-latest",
    instruction="Summarise the customer's message in one sentence. Return only the sentence.",
)


def add_priority(node_input: str):
    urgent = any(word in node_input.lower() for word in ("outage", "is down", "charged twice"))
    return Event(output=f"[{'P1' if urgent else 'P3'}] {node_input}")


def confirm(node_input: str):
    return Event(message=f"Ticket created: {node_input}")


root_agent = Workflow(
    name="ticket_intake",
    edges=[("START", summarise, add_priority, confirm)],
)

The model writes the summary and plain Python decides the priority. You can test add_priority with an ordinary unit test, which you cannot do with a rule buried in a prompt.

Routing with human approval

A router node returns an Event with a route, and the next edge maps each route value to a node. A human-in-the-loop node yields RequestInput: the workflow pauses until a person replies, and the reply becomes the next node's input. The node needs no model, so the pause is deterministic. The example below routes refunds to an approval step, bugs to an acknowledgement and everything else to a retrieval agent, such as the search_docs tool from our RAG agent guide.

from google.adk import Agent, Event, Workflow
from google.adk.events import RequestInput

from .tools import search_docs  # the pgvector tool from the RAG guide


def triage(node_input: str):
    """Route by keywords first; no model call needed for the obvious cases."""
    text = node_input.lower()
    if "refund" in text or "charged twice" in text:
        route = "REFUND"
    elif any(word in text for word in ("error", "crash", "bug")):
        route = "BUG"
    else:
        route = "QUESTION"
    return Event(route=route, output=node_input)


answer = Agent(
    name="answer_question",
    model="gemini-flash-latest",
    instruction="Answer the customer's question from the documentation and cite your sources.",
    tools=[search_docs],
)


def acknowledge_bug(node_input: str):
    return Event(message="Thanks. We have logged this for the on-call engineer and will update you here.")


async def approve_refund(node_input: str):
    yield RequestInput(message=f"Refund request:\n{node_input}\n\nApprove? Reply yes or no.")


def settle_refund(node_input: str):
    if str(node_input).strip().lower().startswith("y"):
        return Event(message="Refund approved. It will reach the original payment method in 3 to 5 business days.")
    return Event(message="Refund declined. A support lead will reply with the reason.")


root_agent = Workflow(
    name="support_workflow",
    edges=[
        ("START", triage),
        (triage, {"REFUND": approve_refund, "BUG": acknowledge_bug, "QUESTION": answer}),
        (approve_refund, settle_refund),
    ],
)

When keywords are not enough, classify with a model instead: an Agent node that replies with a category, followed by a router that turns its reply into a route, as in the ADK docs. If the classifier returns several categories, set route to a list and every matching branch runs. Keep in mind that the router then sees only the category, so keep the original ticket in session state if a branch needs it.

For a yes or no check on a single action inside an LLM agent, ADK also has tool confirmation: mark a function tool as requiring confirmation and the agent pauses before that tool runs. Use a RequestInput node when the approval is a step in your process, and tool confirmation when it guards one risky call.

Fan out with ParallelAgent

When an answer needs facts from independent sources, run the lookups at the same time. Each sub-agent writes its result to session state with output_key, and the next agent reads those keys through {placeholders} in its instruction.

from google.adk.agents import LlmAgent, ParallelAgent, SequentialAgent

policy = LlmAgent(
    name="policy_lookup",
    model="gemini-flash-latest",
    instruction="Find the refund policy that applies to this order. Output only the relevant clause.",
    tools=[search_docs],
    output_key="policy_findings",
)

history = LlmAgent(
    name="order_history",
    model="gemini-flash-latest",
    instruction="Summarise this customer's last three orders. Output only the summary.",
    tools=[get_orders],  # your own function tool
    output_key="order_findings",
)

writer = LlmAgent(
    name="reply_writer",
    model="gemini-flash-latest",
    instruction="""Write the reply to the customer using only these findings.
Policy: {policy_findings}
Orders: {order_findings}""",
)

root_agent = SequentialAgent(
    name="refund_research",
    sub_agents=[ParallelAgent(name="lookups", sub_agents=[policy, history]), writer],
)

Connect tools over MCP

The Model Context Protocol (MCP) is an open standard for exposing tools and data to agents. ADK's McpToolset connects to an MCP server, discovers its tools and hands them to your agent, and tool_filter limits which ones the agent sees. Install the extra with pip install "google-adk[mcp]". The ADK docs also note that agents deployed to production must define McpToolset synchronously in agent.py.

import os

from google.adk.agents import LlmAgent
from google.adk.tools.mcp_tool import McpToolset
from google.adk.tools.mcp_tool.mcp_session_manager import StdioConnectionParams
from mcp import StdioServerParameters

root_agent = LlmAgent(
    name="docs_assistant",
    model="gemini-flash-latest",
    instruction="Help the team find and read internal runbooks.",
    tools=[
        McpToolset(
            connection_params=StdioConnectionParams(
                server_params=StdioServerParameters(
                    command="npx",
                    args=["-y", "@modelcontextprotocol/server-filesystem", os.path.abspath("./runbooks")],
                ),
            ),
            tool_filter=["list_directory", "read_file"],
        )
    ],
)

Every MCP tool definition and raw result enters the agent's context. Past roughly twenty tools, or with large outputs, move the work into a sub-agent wrapped as a tool so the main agent sees only the final answer.

Talk to other agents over A2A

The Agent2Agent (A2A) protocol lets agents built with different frameworks call each other over HTTP. In ADK, to_a2a wraps an agent as an A2A server and generates its agent card, and RemoteA2aAgent lets a local agent use a remote one as a sub-agent. The ADK docs mark the Python A2A integration as experimental, so pin your versions.

# remote side: expose the RAG agent (run with: uvicorn remote:a2a_app --port 8001)
from google.adk.a2a.utils.agent_to_a2a import to_a2a

from support_rag.agent import root_agent

a2a_app = to_a2a(root_agent, port=8001)


# calling side: use it like any other sub-agent
from google.adk.agents import Agent
from google.adk.agents.remote_a2a_agent import AGENT_CARD_WELL_KNOWN_PATH, RemoteA2aAgent

docs_agent = RemoteA2aAgent(
    name="docs_agent",
    description="Answers questions from the support documentation.",
    agent_card=f"http://localhost:8001{AGENT_CARD_WELL_KNOWN_PATH}",
)

root_agent = Agent(
    name="front_desk",
    model="gemini-flash-latest",
    instruction="Delegate documentation questions to docs_agent. Handle greetings yourself.",
    sub_agents=[docs_agent],
)

Upgrading from ADK 1.x

ADK 2.0 is designed to run 1.x agents, but there are breaking changes. Events gain node_info and output fields, so a custom session store that maps events to fixed columns needs a schema change, and strict JSON validators downstream must accept the new fields. BaseAgent now subclasses BaseNode, so custom agents that override 1.x execution methods such as _run_async_impl need rework. Read the 2.0 migration notes before you upgrade a production agent.

A production checklist for multi-agent systems

• Draw the graph before you write prompts. Every branch should have an owner and a test.
• Keep money, permissions and limits in function nodes, not in instructions.
• Give every agent the narrowest tool list that works.
• Persist sessions (Cloud SQL, AlloyDB or Agent Runtime) so a pause for approval survives restarts.
• Evaluate each specialist agent alone, then the whole workflow, and run both in CI.
• Trace every run. The adk web Trace tab is enough in development; export OpenTelemetry traces in production.

RapidStackLab designs and ships multi-agent systems on ADK and LangGraph with human approval steps, evaluation suites and managed operations. Book a free strategy call to map your workflow.