Google ADK 2.0: build a graph-based agent workflow in Python

A support agent with one long prompt and twelve tools looks great in a demo. Then a customer sends a message that needs a fixed sequence: classify it, look up the order, decide, reply. The agent answers, the answer sounds plausible, and you have no way to tell whether it actually looked up the order or invented the status. The control flow lives inside the model's head, so you cannot test it.

Agent Development Kit (ADK) 2.0 from Google adds a graph runtime for exactly this case. You declare nodes and edges, use the model where reasoning is needed, and run plain Python where it is not. What follows builds a support triage workflow, runs it locally, and attaches an eval set so a prompt edit can't quietly break the routing.

Prerequisites

• Python 3.10 or newer
• pip
• A Gemini API key from Google AI Studio
• Comfort with type hints and Pydantic models

Step 1: install and scaffold

python -m venv .venv
source .venv/bin/activate
pip install google-adk
adk create my_agent

adk create writes three files:

my_agent/
    agent.py      # your root agent
    .env          # API keys
    __init__.py

Put the key in my_agent/.env:

echo 'GOOGLE_API_KEY="YOUR_API_KEY"' > my_agent/.env

The only required object in agent.py is root_agent. Everything after this is editing that one file.

Step 2: write tools the model calls correctly

ADK builds the tool schema from your function signature and its docstring. A parameter with a type hint and no default is required. Give it a default and it becomes optional. Parameter descriptions come from the docstring, which makes the docstring prompt engineering rather than documentation.

from google.adk.tools import ToolContext

ORDERS = {
    "A-1042": {"status": "delivered", "total_idr": 149000, "days_since_delivery": 3},
    "A-2098": {"status": "in_transit", "total_idr": 89000, "days_since_delivery": None},
}

def lookup_order(order_id: str, tool_context: ToolContext) -> dict:
    """Looks up one order in the local order table.

    Args:
        order_id (str): Order id in the form A-1042.

    Returns:
        dict: status, total_idr, and days_since_delivery, or status "not_found".
    """
    key = (order_id or "").strip().upper()
    order = ORDERS.get(key)
    if order is None:
        return {"status": "not_found", "order_id": key}
    tool_context.state["temp:last_order_id"] = key
    return {"status": "success", "order_id": key, **order}

Two things are worth copying from that snippet. Returning {"status": "not_found"} instead of raising gives the model something to recover from. And a parameter typed ToolContext gets injected by the framework and hidden from the model, which is how a tool reads session state or triggers an agent transfer.

Variadic parameters (*args, **kwargs) are ignored when ADK generates the schema, so anything the model must supply belongs in an explicit parameter.

Step 3: know where session state lives

session.state is a dictionary, and the key prefix decides its scope:

• no prefix: this conversation only
• user:, as in user:preferred_language, shared across every session for that user
• app: shared across all users of the app
• temp: discarded when the current invocation finishes

Whether any of it survives a restart depends on the session service you pick. InMemorySessionService loses everything, DatabaseSessionService and VertexAiSessionService do not.

State writes inside a tool or a callback go through ToolContext.state or CallbackContext.state and get recorded as events. Writing to a session object you fetched from the session service directly looks fine and then silently fails to persist on a database-backed service. Read there, write here.

You can also inject state into an instruction with braces, for example "Answer in {user:preferred_language}." A missing key raises an error, so use {key?} for values that may not exist yet.

Step 4: define the workflow as a graph

This workflow has four parts: a classifier agent, a router function, two branch agents, and one non-model path for messages that need a human.

from typing import Literal
from pydantic import BaseModel
from google.adk import Agent, Event, Workflow

class TicketIntent(BaseModel):
    """Routing decision for one incoming support message."""
    route: Literal["REFUND", "TRACKING", "OTHER"]
    order_id: str
    reason: str

intent_agent = Agent(
    name="intent_agent",
    model="gemini-2.5-flash",
    instruction=(
        "Read the customer message and classify it as REFUND, TRACKING, or OTHER. "
        "Extract the order id if it appears, otherwise use an empty string. "
        "Keep reason under 20 words."
    ),
    output_schema=TicketIntent,
)

def route(intent: TicketIntent) -> str:
    return intent.route

output_schema makes the model return a validated TicketIntent instead of prose, so the node after it receives a Pydantic object rather than a string you parse by hand. input_schema does the same on the way in.

The branch agents are ordinary Agent objects with the lookup tool attached:

refund_agent = Agent(
    name="refund_agent",
    model="gemini-2.5-flash",
    instruction=(
        "Call lookup_order with the order id before writing anything. "
        "If the order is delivered and fewer than 7 days have passed, approve the refund. "
        "Otherwise explain the policy and offer a manual review. Reply in the customer's language."
    ),
    tools=[lookup_order],
    input_schema=TicketIntent,
)

tracking_agent = Agent(
    name="tracking_agent",
    model="gemini-2.5-flash",
    instruction=(
        "Call lookup_order and report the delivery status in one short paragraph. "
        "If the order is not found, ask for the order id again."
    ),
    tools=[lookup_order],
    input_schema=TicketIntent,
)

def escalate(intent: TicketIntent) -> Event:
    return Event(message=f"Needs a human: {intent.reason}")

Then the graph. START is the entry point, a tuple chains nodes, and a tuple whose second element is a dict turns a node into a router:

root_agent = Workflow(
    name="root_agent",
    edges=[
        ("START", intent_agent, route),
        (route, {
            "REFUND": refund_agent,
            "TRACKING": tracking_agent,
            "OTHER": escalate,
        }),
    ],
)

route returns one of three strings and the dict maps each string to the next node. Nothing in this section asks a model to make the routing decision, so a message that is plainly about delivery status cannot land in the refund path.

One detail to keep in mind: each node's output is the next node's input. The classifier hands a TicketIntent to route, route hands a string key to the branch, and the branch agent ends the run. If you want a fixed step after every branch, add that node to each branch's edge.

Step 5: run it and read the events

adk run my_agent          # terminal chat
adk web --port 8000       # browser UI for development

The web UI is for testing and debugging, and the docs are explicit that it is not a production deployment.

For programmatic access, start the API server and drive it with curl:

adk api_server

curl -X POST http://localhost:8000/apps/my_agent/users/u_123/sessions/s_123 \
  -H "Content-Type: application/json" \
  -d '{"channel": "whatsapp"}'

curl -X POST http://localhost:8000/run \
  -H "Content-Type: application/json" \
  -d '{
    "appName": "my_agent",
    "userId": "u_123",
    "sessionId": "s_123",
    "newMessage": {
      "role": "user",
      "parts": [{"text": "Order A-1042 arrived damaged, I want my money back"}]
    }
  }'

/run returns the full event list at once. /run_sse streams the same events over Server-Sent Events and accepts "streaming": true for token-level output. Read the events, not the final text: a function call event for lookup_order is the proof the lookup actually happened.

Step 6: attach an eval set

Routing bugs come back every time someone edits an instruction. An eval set catches them.

adk eval my_agent tests/triage.test.json --print_detailed_results

{
  "eval_set_id": "triage_refund_path",
  "name": "Refund path",
  "eval_cases": [
    {
      "eval_id": "damaged_order_refund",
      "conversation": [
        {
          "user_content": {
            "parts": [{"text": "Order A-1042 arrived damaged"}],
            "role": "user"
          },
          "final_response": {
            "parts": [{"text": "We have approved the refund for order A-1042."}],
            "role": "model"
          },
          "intermediate_data": {
            "tool_uses": [
              {"name": "lookup_order", "args": {"order_id": "A-1042"}}
            ],
            "intermediate_responses": []
          }
        }
      ],
      "session_input": {
        "app_name": "my_agent",
        "user_id": "test_user",
        "state": {}
      }
    }
  ]
}

The tool_uses list is the trajectory you expect, and ADK compares it against what the agent actually did. That is the check a plain assertion on the final text can't make. The test file schema is Pydantic-backed, and the docs cover AgentEvaluator.evaluate for running the same set inside pytest or a CI job.

Step 7: deploy

adk deploy docker --with_ui my_agent
adk deploy cloud_run --with_ui my_agent

The Docker path gives you a container you can host anywhere, and Cloud Run is the managed option. Drop --with_ui for anything public.

When a graph is the right shape

A single agent with tools is still correct when the order of steps doesn't matter and the model can choose freely. Template workflows (SequentialAgent, LoopAgent, ParallelAgent) cover fan-out and loops over sub-agents. In ADK 2.0 the graph runtime handles those cases with more control, and the docs mark template workflows as superseded for Python and Go. On TypeScript, Java, or Kotlin, template workflows remain the supported path.

Reach for a graph when the route between steps has to be predictable, when you need code between model calls, or when a node should stop and wait for a human. Keep the model out of any decision you already know the answer to.

Next steps

• Wrap intent_agent in an AgentTool if another agent needs it, so the routing logic has one copy.
• Swap InMemorySessionService for the database-backed service before anyone but you touches the workflow.
• Add a callback to ship the event list into your tracing stack, since the events are already there.

References

• Graph-based agent workflows, ADK docs
• State: the session's scratchpad
• Why evaluate agents
• google/adk-python on GitHub