The Google ADK Masterclass

Google released the Agent Development Kit (ADK) at Cloud Next 2025 as the same open-source toolkit that powers agents inside Google's own products. It is a code-first, Python-native framework for building, testing, and shipping agentic systems — single agents, hierarchies of specialists, or tool-using workflows. Java and TypeScript ports exist; this guide focuses on Python because that is where the API is most complete.

Three commitments define ADK and explain most of its design choices:

• Code-first. Agents are Python objects, not YAML or graphs in a UI. You compose, test, and debug with normal language tools.
• Model-agnostic. Gemini is the default and the deepest integration, but ADK runs on Anthropic, OpenAI, Mistral, Ollama, and anything reachable through LiteLLM.
• Deployment-ready. The same agent runs locally in a dev loop, in a container on Cloud Run, or on the managed Vertex AI Agent Engine with one command.

The runtime, in one picture

Every ADK app has the same shape. A Runner orchestrates a turn: it receives a user message, hands it to the root Agent, lets the agent call the LLM and tools, persists everything to a Session, and emits events you stream to the user.

Where ADK fits next to other frameworks

• vs. LangGraph. Both model agentic flows. LangGraph is graph-first; ADK is class-first with workflow primitives (Sequential / Parallel / Loop) when you need explicit orchestration. ADK ships an opinionated runtime, a CLI, an eval framework, and Vertex deployment in the box.
• vs. CrewAI. Both model role-based collaborations. ADK is closer to the metal: you can build the same patterns but you also have callbacks, streaming, MCP, and direct LLM control.
• vs. AutoGen. AutoGen optimizes for emergent conversation between agents; ADK leans toward predictable, testable hierarchies with explicit delegation rules.

Install and authenticate

pip install google-adk

# Use either the Gemini Developer API (free tier, no GCP needed)…
export GOOGLE_API_KEY="your-gemini-api-key"
export GOOGLE_GENAI_USE_VERTEXAI=False

# …or Vertex AI (production GCP).
export GOOGLE_CLOUD_PROJECT="your-project-id"
export GOOGLE_CLOUD_LOCATION="us-central1"
export GOOGLE_GENAI_USE_VERTEXAI=True

A working agent fits in 12 lines. The agent has a model, a description, an instruction (its system prompt), and zero or more tools.

from google.adk.agents import Agent

def get_weather(city: str) -> dict:
    """Returns the current weather for a given city."""
    # Replace with a real API call in production.
    return {"city": city, "condition": "sunny", "temp_c": 24}

root_agent = Agent(
    name="weather_agent",
    model="gemini-2.5-flash",
    description="A friendly assistant that answers weather questions.",
    instruction="Use the get_weather tool when the user asks about weather. Be concise.",
    tools=[get_weather],
)

Run it locally with one CLI command

Save the agent in weather_agent/agent.py (the folder name becomes the app name) and point the dev UI at the parent directory:

# From the parent of weather_agent/
adk web

# Or a terminal REPL:
adk run weather_agent

adk web opens a browser-based playground at http://localhost:8000 with chat, event traces, and a session inspector. adk run is a stdin/stdout loop for fast iteration.

Programmatic invocation

For tests, jobs, or anything embedded, you drive the agent yourself through Runner:

import asyncio
from google.adk.runners import Runner
from google.adk.sessions import InMemorySessionService
from google.genai import types

APP, USER, SESSION = "weather_app", "u1", "s1"

session_service = InMemorySessionService()
runner = Runner(agent=root_agent, app_name=APP, session_service=session_service)

async def main():
    await session_service.create_session(app_name=APP, user_id=USER, session_id=SESSION)
    message = types.Content(role="user", parts=[types.Part(text="Weather in Tokyo?")])

    async for event in runner.run_async(user_id=USER, session_id=SESSION, new_message=message):
        if event.is_final_response():
            print(event.content.parts[0].text)

asyncio.run(main())

Every step the agent takes — model call, tool call, tool response, final answer — surfaces as an Event. The same event stream powers the web UI, the eval framework, and your production observability.

A tool is anything the LLM can call to affect the world or fetch information. ADK gives you five flavors and they all plug into the same tools=[...] list.

1. Python functions (the default)

def book_flight(origin: str, destination: str, date: str) -> dict:
    """Books a one-way flight. Date must be in YYYY-MM-DD format."""
    # call your booking API
    return {"confirmation": "ABC123", "price_usd": 412}

Add it to tools=[book_flight] and the model can call it. Use Pydantic models as parameters when you need richer schemas.

2. Built-in Google tools

from google.adk.tools import google_search, built_in_code_execution

agent = Agent(
    name="research_agent",
    model="gemini-2.5-pro",
    instruction="Use google_search for facts and code_execution for math.",
    tools=[google_search, built_in_code_execution],
)

These are first-party Google tools that run server-side inside the Gemini call — no extra latency, no extra auth.

3. MCP (Model Context Protocol) servers

Any MCP server — official Neo4j, GitHub, filesystem, your own — drops in as a toolset:

from google.adk.tools.mcp_tool import MCPToolset, StdioServerParameters

# Spin up a local MCP server as a subprocess and expose its tools.
mcp_toolset = MCPToolset(
    connection_params=StdioServerParameters(
        command="npx",
        args=["-y", "@modelcontextprotocol/server-filesystem", "/workspace"],
    ),
)

agent = Agent(
    name="fs_agent",
    model="gemini-2.5-flash",
    instruction="Help the user read and write files in /workspace.",
    tools=[mcp_toolset],
)

4. OpenAPI specs

from google.adk.tools.openapi_tool import OpenAPIToolset

openapi_tools = OpenAPIToolset(spec_str=open("petstore.yaml").read(), spec_str_type="yaml")
agent = Agent(..., tools=[openapi_tools])

Every operation in the spec becomes a callable tool with the right schema.

5. AgentTool: an agent as a tool

Wrap a sub-agent so its parent can call it like any other tool. This is the core of delegation:

from google.adk.tools import AgentTool

translator = Agent(name="translator", model="gemini-2.5-flash",
                   instruction="Translate text to French.")

coordinator = Agent(
    name="coordinator",
    model="gemini-2.5-pro",
    tools=[AgentTool(agent=translator)],
)

Reading and writing state from inside a tool

Declare a tool_context: ToolContext parameter and ADK injects the session bag. You can read state, mutate it, save artifacts, or escape to the framework.

from google.adk.tools import ToolContext

def remember_preference(key: str, value: str, tool_context: ToolContext) -> str:
    """Saves a user preference for the rest of the session."""
    tool_context.state[f"user:{key}"] = value
    return f"Got it — I'll remember {key} = {value}."

ADK separates three timescales of context, and they have three different services:

• Session — one conversation, including its event history.
• State — a key/value bag attached to a session, your scratchpad for the current task.
• Memory — long-term facts that survive across sessions for the same user.

SessionService

from google.adk.sessions import InMemorySessionService
# Dev / tests.
session_service = InMemorySessionService()

# Production (persists to a SQL database):
from google.adk.sessions import DatabaseSessionService
session_service = DatabaseSessionService(db_url="postgresql+psycopg://user:pwd@host/db")

# Or fully managed:
from google.adk.sessions import VertexAiSessionService
session_service = VertexAiSessionService(project="my-project", location="us-central1")

You pick the service once at Runner construction; everything else stays the same.

State and its prefixes

State is just a dict, but key prefixes give it three lifetimes:

# Session-only (default): cleared when the session ends.
state["draft"] = "..."

# User-scoped: persists across sessions for the same user.
state["user:preferred_language"] = "fr"

# App-scoped: shared by every user of this app.
state["app:rate_limit_qps"] = 5

# Ephemeral: never persisted, only lives for this turn.
state["temp:tool_choice"] = "search"

The prefix is the contract — the session service decides where each kind lands.

Memory for cross-session recall

from google.adk.memory import InMemoryMemoryService
from google.adk.tools import load_memory  # built-in tool

memory_service = InMemoryMemoryService()

agent = Agent(
    name="assistant",
    model="gemini-2.5-pro",
    instruction="Use load_memory when the user references something from a past conversation.",
    tools=[load_memory],
)

runner = Runner(
    agent=agent,
    app_name="assistant_app",
    session_service=session_service,
    memory_service=memory_service,
)

On Vertex AI you swap in VertexAiMemoryBankService and get a managed, vector-indexed memory backend that learns continuously from past sessions.

Most non-trivial agents are actually several agents. ADK gives you two composition styles. LLM-driven delegation uses the model's judgement to pick a sub-agent. Workflow agents impose deterministic orchestration when you don't want a model deciding.

Hierarchical delegation

Add specialists to sub_agents. ADK injects their names + descriptions into the parent's prompt and exposes a transfer_to_agent tool. The parent decides when to hand off.

researcher = Agent(name="researcher", model="gemini-2.5-flash",
                   description="Finds facts and citations using google_search.",
                   tools=[google_search])

writer = Agent(name="writer", model="gemini-2.5-pro",
               description="Turns research notes into polished prose.")

reviewer = Agent(name="reviewer", model="gemini-2.5-flash",
                 description="Critiques the writer's output for accuracy.")

coordinator = Agent(
    name="article_coordinator",
    model="gemini-2.5-pro",
    instruction=(
        "Plan the article, delegate research to 'researcher', drafting to 'writer', "
        "and quality control to 'reviewer'. Loop until reviewer approves."
    ),
    sub_agents=[researcher, writer, reviewer],
)

Workflow agents: when you want determinism

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

# Run in order — each step sees the previous output via state.
pipeline = SequentialAgent(
    name="research_pipeline",
    sub_agents=[fetch_sources_agent, summarize_agent, format_report_agent],
)

# Fan out, then merge. Each branch runs concurrently.
fanout = ParallelAgent(
    name="multi_source",
    sub_agents=[news_agent, wikipedia_agent, papers_agent],
)

# Repeat until a sub-agent calls tool_context.actions.escalate.
refine = LoopAgent(
    name="refine_until_good",
    sub_agents=[draft_agent, critic_agent],
    max_iterations=5,
)

Mixing the two

Workflow agents are agents, so you can nest. A common production shape: a top-level SequentialAgent with deterministic stages, and one stage is an LLM-driven coordinator that delegates internally.

app = SequentialAgent(
    name="content_pipeline",
    sub_agents=[
        intake_classifier,      # LlmAgent: routes by topic
        researcher_coordinator, # LlmAgent with sub_agents=[…]
        ParallelAgent(name="qc",
                      sub_agents=[fact_check_agent, style_check_agent]),
        publisher_agent,
    ],
)

Callbacks are hooks ADK invokes around every important transition. Return None to let execution proceed; return a value to short-circuit and substitute that value as the result.

The hook points are symmetric:

• before_agent_callback / after_agent_callback
• before_model_callback / after_model_callback
• before_tool_callback / after_tool_callback

Block a tool call based on policy

from google.adk.agents.callback_context import CallbackContext
from google.adk.tools.base_tool import BaseTool
from google.adk.tools.tool_context import ToolContext
from typing import Optional

BLOCKED = {"send_payment", "delete_user"}

def policy_gate(tool: BaseTool, args: dict, tool_context: ToolContext) -> Optional[dict]:
    if tool.name in BLOCKED and not tool_context.state.get("user:is_admin"):
        return {"error": "policy_denied", "tool": tool.name}
    return None  # allow

agent = Agent(
    name="safe_assistant",
    model="gemini-2.5-flash",
    tools=[...],
    before_tool_callback=policy_gate,
)

Redact PII from the model's input

import re
from google.adk.models.llm_request import LlmRequest

EMAIL = re.compile(r"[\w.+-]+@[\w-]+\.[\w.-]+")

def redact_pii(callback_context: CallbackContext, llm_request: LlmRequest):
    for content in llm_request.contents:
        for part in (content.parts or []):
            if part.text:
                part.text = EMAIL.sub("[email-redacted]", part.text)
    return None

agent = Agent(..., before_model_callback=redact_pii)

ADK ships a first-party evaluation framework. You write a few example interactions, declare what good looks like, and run them in CI.

The eval set format

An eval set is a JSON file with cases. Each case lists the conversation and the expected tool trajectory + final response.

{
  "eval_set_id": "weather.evalset.json",
  "eval_cases": [
    {
      "eval_id": "tokyo_weather",
      "conversation": [
        {
          "user_content": { "parts": [{ "text": "What's the weather in Tokyo?" }] },
          "final_response": { "parts": [{ "text": "It's currently sunny in Tokyo, around 24 °C." }] },
          "intermediate_data": {
            "tool_uses": [
              { "name": "get_weather", "args": { "city": "Tokyo" } }
            ]
          }
        }
      ]
    }
  ]
}

Run from the CLI or from Python

adk eval weather_agent weather.evalset.json \
  --config_file_path eval_config.json \
  --print_detailed_results

import pytest
from google.adk.evaluation import AgentEvaluator

@pytest.mark.asyncio
async def test_weather_agent():
    await AgentEvaluator.evaluate(
        agent_module="weather_agent",
        eval_dataset_file_path_or_dir="weather.evalset.json",
    )

Two default metrics: tool_trajectory_avg_score (did the agent call the right tools with the right args, in the right order?) and response_match_score (semantic match against the expected final answer). You can plug in your own.

The same agent code runs in three places without modification. The deployment target only changes who runs the Runner.

Local development

adk web              # browser playground + event traces
adk run weather_agent # terminal REPL

Vertex AI Agent Engine (managed)

Agent Engine is the production target for ADK agents. It handles scaling, sessions, monitoring, and identity. Deploy with one command:

adk deploy agent_engine weather_agent \
  --project=$GOOGLE_CLOUD_PROJECT \
  --region=us-central1 \
  --staging_bucket=gs://my-staging

You get a stable endpoint with built-in session persistence, identity-aware tools, and OpenTelemetry exports.

Cloud Run (serverless containers)

adk deploy cloud_run weather_agent \
  --project=$GOOGLE_CLOUD_PROJECT \
  --region=us-central1 \
  --service_name=weather-agent

The CLI builds an OCI image (with FastAPI serving the agent), pushes to Artifact Registry, and deploys.

Your own container

For GKE, EKS, or on-prem, point at the same Dockerfile and run adk api_server:

FROM python:3.12-slim
WORKDIR /app
COPY . .
RUN pip install --no-cache-dir google-adk
EXPOSE 8000
CMD ["adk", "api_server", "--port", "8000", "."]

Project layout that ADK expects

my_project/
├── weather_agent/
│   ├── __init__.py     # from . import agent
│   ├── agent.py        # defines root_agent
│   └── .env            # GOOGLE_API_KEY=..., or Vertex envs
├── tests/
│   └── test_weather.py
└── pyproject.toml

agent.py must expose a module-level root_agent. The folder name (weather_agent) becomes the app name in the CLI and the URL slug in adk web.

Common configuration knobs

from google.genai import types

Agent(
    name="precise_agent",
    model="gemini-2.5-pro",
    instruction="...",
    tools=[...],
    generate_content_config=types.GenerateContentConfig(
        temperature=0.1,
        max_output_tokens=2048,
        top_p=0.95,
    ),
    output_schema=None,    # Pydantic model => model returns structured JSON
    output_key="result",   # store final output under state["result"]
    disallow_transfer_to_parent=False,
    disallow_transfer_to_peers=False,
)

Streaming events to your own UI

async for event in runner.run_async(user_id=u, session_id=s, new_message=msg):
    if event.is_partial:                    # token stream
        yield {"type": "delta", "text": event.content.parts[0].text}
    elif event.get_function_calls():        # tool invocation
        for call in event.get_function_calls():
            yield {"type": "tool_call", "name": call.name, "args": call.args}
    elif event.is_final_response():
        yield {"type": "final", "text": event.content.parts[0].text}

Where to read next

• google.github.io/adk-docs — the official documentation.
• github.com/google/adk-python — source, samples, and the issue tracker.
• github.com/google/adk-samples — production-shaped reference agents (research, customer-service, RAG, multi-agent).
• MCP Toolbox for Databases — declarative, pre-validated DB tools with native ADK integration; pairs naturally with the Neo4j tutorial.