Google Agent Development Kit (ADK) Handbook

Getting Started — Install & Setup

Google Agent Development Kit (ADK) is an open-source, code-first framework for building, evaluating, and deploying agentic AI applications. ADK supports Python, Go, Java, and TypeScript; this handbook focuses on Python, with Go equivalents noted where the graph API differs.

Installation

# Install ADK (Python 3.10+; ADK 2.0 core requires 3.10+, some extras want 3.11+)
pip install google-adk        # currently 2.3.x — release cadence is roughly bi-weekly

# Optional extras — the 2.0 line expanded this list significantly
pip install google-adk[eval]           # evaluation tools (adk eval)
pip install google-adk[a2a]            # Agent-to-Agent protocol client/server
pip install google-adk[mcp]            # Model Context Protocol toolsets
pip install google-adk[toolbox]        # pre-built connectors: GitHub, Jira, Notion, MongoDB, Qdrant...
pip install google-adk[agent-identity] # workload identity / auth for agent-to-service calls
pip install google-adk[gcp]            # Vertex AI / Agent Runtime deployment support
pip install google-adk[all]            # everything, incl. litellm, otel-gcp, db, slack, e2b

# Create a new agent project scaffold
adk create my_agent

Project Structure

The adk create command scaffolds a minimal but complete agent project. Since 2.0, the scaffold can generate either an Agent-first project or a Workflow-first (graph) project:

my_agent/
  agent.py          # root_agent definition — required entry point
  .env              # API keys and environment variables
  __init__.py       # exposes root_agent for the ADK runner

The only required element is a root_agent variable in agent.py. It can be a plain Agent, or — since 2.0 — a Workflow instance composed of nodes and edges. Everything else is optional.

First Agent

Update agent.py with a working agent that uses a tool:

from google.adk import Agent

# Tool: plain Python function — ADK infers schema from type hints + docstring
def get_weather(city: str) -> dict:
    """Returns current weather conditions for a city.

    Args:
        city: The name of the city to get weather for.

    Returns:
        A dict with status, city, and temperature fields.
    """
    # Replace with a real weather API call in production
    return {"status": "success", "city": city, "temp_c": 22}

root_agent = Agent(
    name="weather_agent",
    model="gemini-3-flash",
    description="Provides current weather for any city.",
    instruction="You are a weather assistant. Use get_weather to answer questions.",
    tools=[get_weather],
)

Running Your Agent

adk run my_agent

adk web path/to/agents_dir --port 8000

Environment Configuration

# Gemini API (Google AI Studio)
GOOGLE_API_KEY="your-key-here"

# OR: Gemini Enterprise Agent Platform (formerly Vertex AI)
GOOGLE_CLOUD_PROJECT="your-project-id"
GOOGLE_CLOUD_LOCATION="us-central1"

# Default-on OpenTelemetry tracing for agents deployed on Agent Runtime — no extra config needed
# For local/self-hosted OTel export instead:
OTEL_EXPORTER_OTLP_ENDPOINT="http://localhost:4317"

Build Your Agent

The primary agent type in ADK is LlmAgent (aliased as Agent). It uses a large language model as its core engine to reason, plan, use tools, and generate responses. In ADK 2.0, an Agent can also run standalone or as a node inside a Workflow graph — the same class powers both models.

LlmAgent Parameters

Writing Effective Instructions

Instructions are the single biggest lever on agent quality. Treat them as a precision contract.

root_agent = Agent(
    name="support_agent",
    model="gemini-3-flash",
    instruction="""
You are a Tier-1 support agent for AcmeCorp SaaS.

SCOPE
- Answer questions about billing, account access, and product features.
- For engineering or outages, route the user to the escalation_agent.
- Never discuss competitor products or internal roadmaps.

FORMAT
- Keep responses under 120 words.
- Use bullet points for steps.
- End every response with: 'Is there anything else I can help with?'

TOOLS
- Use search_kb before answering any product feature question.
- Use get_account_info only when the user provides their account ID.
    """,
    tools=[search_kb, get_account_info],
    sub_agents=[escalation_agent],
)

Supported Models

# Any non-Gemini model via LiteLLM
from google.adk.models.lite_llm import LiteLlm

agent = Agent(
    model=LiteLlm(model="anthropic/claude-sonnet-4-6"),
    ...
)

# Local Ollama
agent = Agent(
    model=LiteLlm(model="ollama/mistral"),
    ...
)

Sessions & Memory

ADK separates sessions (short-term per-conversation state) from memory (long-term cross-session recall). On Google Cloud, the managed memory service was renamed as part of the June 2026 Agent Platform rebrand.

from google.adk.runners import Runner
from google.adk.sessions import InMemorySessionService
from google.adk.memory import InMemoryMemoryService

runner = Runner(
    agent=root_agent,
    app_name="my_app",
    session_service=InMemorySessionService(),
    memory_service=InMemoryMemoryService(),   # optional
)

# Create a session and run
session = await runner.session_service.create_session(
    app_name="my_app", user_id="user-123"
)
async for event in runner.run_async(
    user_id="user-123",
    session_id=session.id,
    new_message=Content(parts=[Part(text="What is the weather in Tokyo?")]),
):
    if event.is_final_response():
        print(event.content.parts[0].text)

Agent Teams & A2A

ADK's Agent Teams pattern is a hierarchical multi-agent primitive: a root orchestrator delegates tasks to specialised sub-agents. The orchestrator is an LlmAgent with sub_agents; it decides autonomously which sub-agent handles each task. For deterministic, auditable delegation, ADK 2.0 layers the Task API and a formalised A2A protocol on top of this same model.

Defining an Agent Team

from google.adk import Agent

# Specialist agents
billing_agent = Agent(
    name="billing_agent",
    model="gemini-3-flash",
    description="Handles billing questions, invoices, and subscription changes.",
    instruction="You are a billing specialist. Only answer billing-related questions.",
    tools=[get_invoice, update_subscription, apply_credit],
)

technical_agent = Agent(
    name="technical_agent",
    model="gemini-3-pro",  # richer model for complex debugging
    description="Handles technical support, bugs, and integration issues.",
    instruction="You are a technical support engineer. Diagnose and resolve technical issues.",
    tools=[search_kb, run_diagnostics, create_ticket],
)

# Orchestrator — delegates via sub_agents, no direct tools needed
root_agent = Agent(
    name="support_orchestrator",
    model="gemini-3-flash",
    description="Routes customer support requests to the right specialist.",
    instruction="""
You are a customer support orchestrator.
Analyse the user's intent and delegate to the appropriate specialist agent.
- Billing questions → billing_agent
- Technical issues  → technical_agent
Never answer domain questions yourself — always delegate.
    """,
    sub_agents=[billing_agent, technical_agent],
)

A2A Protocol — Formalised in 2.0

Agent-to-Agent (A2A) delegation lets one agent call a "specialist" agent running in a different process, region, or even a different framework entirely, discovered via a URL endpoint. ADK 2.0 formalises this with the a2a extra, structured task hand-off, and a shared Task API so that a delegation can be:

from google.adk.a2a import A2AClient
from google.adk import Agent

# Delegate to a specialist agent running as its own remote service
credit_check_agent = A2AClient(
    name="credit_check_agent",
    endpoint="https://risk.internal.acme.com/agents/credit-check",
    description="Runs a credit check and returns an approval decision.",
)

root_agent = Agent(
    name="loan_intake_agent",
    model="gemini-3-flash",
    instruction="Collect loan application details, then delegate credit checks to credit_check_agent.",
    sub_agents=[credit_check_agent],
)

Agent Registry

Context Sharing Between Agents

All agents in a hierarchical team share the same session state. Sub-agents can read and write state keys to pass results back to the orchestrator.

from google.adk.tools import ToolContext

def fetch_user_profile(user_id: str, tool_context: ToolContext) -> dict:
    """Fetch and cache user profile in session state."""
    profile = call_user_api(user_id)
    tool_context.state["user_profile"] = profile
    return profile

# In another agent's tool, read state written by a sibling agent:
def personalize_response(tool_context: ToolContext) -> str:
    profile = tool_context.state.get("user_profile", {})
    return f"Hello {profile.get('name', 'there')}!"

Multi-Tool Agents

A multi-tool agent is an LlmAgent equipped with several tools. The LLM autonomously decides which tools to call, in what order, and how to chain their outputs to fulfil the user's request.

Tool Types Overview

Multi-Tool Agent Example

from google.adk import Agent
from google.adk.tools import google_search, code_execution

def query_database(sql: str) -> dict:
    """Run a read-only SQL query against the analytics database.

    Args:
        sql: A valid SELECT statement.

    Returns:
        A dict with 'rows' (list of row dicts) and 'row_count'.
    """
    results = db.execute(sql).fetchall()
    return {"rows": [dict(r) for r in results], "row_count": len(results)}

analyst_agent = Agent(
    name="analyst_agent",
    model="gemini-3-pro",
    instruction="""
You are a data analyst. You can query the database, search the web for context,
execute Python for data transformation, and email summaries.
Always validate your SQL is read-only before executing.
    """,
    tools=[
        query_database,
        send_email,
        google_search(),     # built-in: web grounding
        code_execution(),    # built-in: run Python for analysis
    ],
)

Tool Result Handling

ADK serialises tool return values to JSON and injects them back into the LLM context as tool response parts. Best practices:

• Return structured dicts, not raw strings — the LLM reasons better over structured data.
• Always include a status key ("success" or "error") so the LLM can handle failures gracefully.
• Return only what the LLM needs — large payloads inflate token usage and can confuse reasoning.
• Use consistent key names across tools to reduce LLM confusion (e.g. always "items" not sometimes "results").

Workflow Runtime — Graph-Based Orchestration

ADK 2.0's headline change is the Workflow Runtime: a graph-based execution engine that transitions ADK from a purely hierarchical agent executor into a system where Agents, Tools, and plain Python Functions are all evaluated as nodes in a directed graph connected by edges. This gives you deterministic, LLM-free control flow with conditional routing, fan-out/fan-in, retries, loops, and human-in-the-loop pauses — all as first-class, resumable graph behaviour.

Nodes and Edges — The Mental Model

A node is any unit of work: an Agent (LLM call), a FunctionNode (pure Python — free and instant), or a JoinNode (waits for parallel branches to complete). Edges describe the flow between nodes, and can be unconditional or conditional on a node's output.

Basic Sequential Graph

from google.adk import Agent, Workflow

generate_fruit_agent = Agent(
    name="generate_fruit_agent",
    instruction="Return the name of a random fruit. Return only the name.",
)
generate_benefit_agent = Agent(
    name="generate_benefit_agent",
    instruction="Tell me a health benefit about the specified fruit.",
)

# A 3-tuple edge chains nodes in order: START → A → B
root_agent = Workflow(
    name="root_agent",
    edges=[("START", generate_fruit_agent, generate_benefit_agent)],
)

Conditional Routing

A router node returns a route key; the matching edge decides where execution goes next. This is a plain Python function or a lightweight agent — no LLM call required if the logic is deterministic.

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

def router(node_input: str) -> Event:
    """Route to task B or C based on node_input."""
    if is_urgent(node_input):
        return Event(route="RUN_TASK_C")
    return Event(route="RUN_TASK_B")

task_B_agent = Agent(name="task_B_agent", model="gemini-3-flash", instruction="Handle a standard request.")

def task_C_node(node_input: str) -> Event:
    """A FunctionNode to execute node C — zero LLM cost."""
    return Event(output="Escalated to on-call engineer.")

root_agent = Workflow(
    name="routing_workflow",
    edges=[
        ("START", task_A_node, router),
        (router, {
            "RUN_TASK_B": task_B_agent,
            "RUN_TASK_C": task_C_node,
        }),
    ],
)

Fan-Out / Fan-In with JoinNode

Run several branches in parallel from a shared source, then wait for all of them to complete before continuing — the graph equivalent of ParallelAgent, but composable with conditional routing and retries.

from google.adk import Workflow
from google.adk.workflow import JoinNode

my_join_node = JoinNode(name="my_join_node")

root_agent = Workflow(
    name="parallel_research",
    edges=[
        ("START", parallel_task_A, my_join_node),
        ("START", parallel_task_B, my_join_node),
        ("START", parallel_task_C, my_join_node),
        (my_join_node, final_summary_agent),
    ],
)

Retries, Timeouts & Human-in-the-Loop

Every node accepts a retry_config and timeout, and any node can pause the graph to wait for human input — the same mechanism the Task API uses for multi-turn conversations.

from google.adk.workflow import RetryConfig

flaky_api_node = FunctionNode(
    name="call_flaky_partner_api",
    fn=call_partner_api,
    retry_config=RetryConfig(
        max_attempts=3,
        initial_delay=1.0,
        backoff_factor=2.0,
        jitter=True,
        exceptions=(ConnectionError, TimeoutError),
    ),
    timeout=30,   # hard wall-clock cap in seconds
)

approval_gate = Agent(
    name="approval_gate",
    mode="task",
    instruction="Summarise the pending action and wait for a human 'approve' or 'reject'.",
)  # pauses the graph until a human responds via the ADK Web UI or API

Choosing an Orchestration Style

Agent & Graph Routing

Routing controls which agent — or which graph node — handles a given request. ADK supports everything from fully LLM-driven delegation to deterministic, code-based routing inside a Workflow.

LLM-Based Routing (Default, Hierarchical)

The orchestrator's LLM reads the user input and selects a sub-agent based on each agent's description. No routing code required.

router = Agent(
    name="router",
    model="gemini-3-flash",
    instruction="Route the user's request to the most appropriate specialist.",
    sub_agents=[billing_agent, tech_agent, onboarding_agent],
    # ADK uses agent.description to build the routing prompt automatically
)

LLM-as-Router in a Graph

For graph-based flows, an LlmAgent can act purely as a classifier: it emits a route tag, and a trivial function node maps that tag to an edge. The model decides; the graph makes the decision reliable, observable, and resumable.

classifier = Agent(
    name="classifier", model="gemini-3.1-flash-lite",
    instruction="Classify the message as one of: question, exclamation, statement. Reply with only that word.",
)

def emit_route(classifier_output: str) -> Event:
    return Event(route=classifier_output.strip().lower())

root_agent = Workflow(
    name="llm_router_workflow",
    edges=[
        ("START", classifier, emit_route),
        (emit_route, {
            "question":    question_handler,
            "exclamation": exclamation_handler,
            "statement":   statement_handler,
        }),
    ],
)

Custom Router Function (Hierarchical)

For deterministic or cost-optimised routing outside a graph, provide a before_agent_callback that sets the target agent name in state:

from google.adk.agents.callback_context import CallbackContext
import re

def intent_router(callback_context: CallbackContext):
    """Keyword-based intent classification before LLM routing."""
    user_msg = callback_context.user_content.parts[0].text.lower()

    if re.search(r"invoice|billing|payment|refund", user_msg):
        callback_context.state["intended_agent"] = "billing_agent"
    elif re.search(r"error|crash|bug|integrate", user_msg):
        callback_context.state["intended_agent"] = "tech_agent"
    # Returning None lets the LLM orchestrator handle ambiguous cases

Routing Patterns Summary

Agent Configuration

ADK exposes granular configuration through GenerateContentConfig for LLM parameters, RunConfig for runtime execution settings, and (2.0+) RetryConfig/timeouts at the node level.

LLM Generation Config

from google.genai.types import GenerateContentConfig, SafetySetting, HarmCategory, HarmBlockThreshold

agent = Agent(
    name="configured_agent",
    model="gemini-3-flash",
    instruction="You are a precise data extraction assistant.",
    generate_content_config=GenerateContentConfig(
        temperature=0.2,          # low = more deterministic
        top_p=0.9,
        max_output_tokens=2048,
        thinking_level="low",   # "minimal" | "low" | "medium" | "high" — Gemini 3.x thinking control
        stop_sequences=["END_OF_OUTPUT"],
        safety_settings=[
            SafetySetting(
                category=HarmCategory.HARM_CATEGORY_DANGEROUS_CONTENT,
                threshold=HarmBlockThreshold.BLOCK_ONLY_HIGH,
            )
        ],
    ),
)

Runtime Config

from google.adk.runners import RunConfig

run_config = RunConfig(
    streaming_mode="SSE",        # "SSE" | "BIDI" | None
    max_llm_calls=20,              # circuit breaker: max LLM calls per run
    response_modalities=["TEXT"],  # "TEXT" | "AUDIO" | "IMAGE"
)

async for event in runner.run_async(
    user_id="u1", session_id="s1",
    new_message=msg,
    run_config=run_config,
):
    ...

Key Configuration Parameters

Structured Output

Force the LLM to return JSON matching a Pydantic schema:

from pydantic import BaseModel
from google.adk import Agent

class ExtractedOrder(BaseModel):
    order_id: str
    customer_name: str
    total: float
    line_items: list[str]

extractor = Agent(
    name="order_extractor",
    model="gemini-3-flash",
    instruction="Extract order details from the provided email text.",
    output_schema=ExtractedOrder,  # response is always valid JSON
)

Custom Agents & Nodes

Custom agents are created by subclassing BaseAgent. Use them when you need unique operational logic, specific control flows, or integrations not covered by LlmAgent or the workflow primitives. In graph-first designs, the equivalent extension point is a FunctionNode or a custom BaseNode subclass.

BaseAgent Contract (Hierarchical)

from google.adk.agents import BaseAgent
from google.adk.agents.invocation_context import InvocationContext
from google.adk.events import Event
from typing import AsyncGenerator

class MyCustomAgent(BaseAgent):
    """A custom agent with bespoke control flow."""

    validator: BaseAgent
    executor: BaseAgent
    max_retries: int = 3

    async def _run_async_impl(
        self, ctx: InvocationContext
    ) -> AsyncGenerator[Event, None]:
        """Core execution logic — must yield Event objects."""

        for attempt in range(self.max_retries):
            async for event in self.validator.run_async(ctx):
                yield event

            if ctx.session.state.get("validation_passed"):
                async for event in self.executor.run_async(ctx):
                    yield event
                return

            ctx.session.state["retry_count"] = attempt + 1

        ctx.session.state["failed"] = True

Custom FunctionNode (Graph)

Any pure Python function can be dropped directly into a Workflow's edges list. Use a FunctionNode wrapper when you need to attach a retry_config, timeout, or explicit schema:

from google.adk.workflow import FunctionNode, RetryConfig

def validate_and_normalize(raw_input: dict) -> dict:
    """Deterministic validation node — no LLM, no cost."""
    if not raw_input.get("email"):
        raise ValueError("Missing email field")  # propagates to RetryConfig, don't swallow it
    return {**raw_input, "email": raw_input["email"].strip().lower()}

validate_node = FunctionNode(
    name="validate_and_normalize",
    fn=validate_and_normalize,
    retry_config=RetryConfig(max_attempts=2),
    timeout=5,
)

Custom Agent / Node Use Cases

Guard Agent Example

from google.adk.agents import BaseAgent

class InputGuardAgent(BaseAgent):
    """Validates inputs before passing to inner_agent."""
    inner_agent: BaseAgent
    forbidden_patterns: list[str] = []

    async def _run_async_impl(self, ctx):
        user_text = ctx.user_content.parts[0].text if ctx.user_content else ""

        for pattern in self.forbidden_patterns:
            if pattern.lower() in user_text.lower():
                yield self._make_final_response(
                    "I'm not able to help with that request."
                )
                return

        async for event in self.inner_agent.run_async(ctx):
            yield event

Custom Tools & Integrations

Custom tools are Python callables that agents can invoke. ADK auto-generates the tool schema from type annotations and the Google-style docstring. No additional registration boilerplate is needed.

Function Tool (Recommended)

def search_documents(
    query: str,
    collection: str = "general",
    max_results: int = 5,
) -> dict:
    """Search an internal document collection.

    Use this tool to find relevant documents, policies, or knowledge-base
    articles matching a natural-language query.

    Args:
        query: Natural-language search query.
        collection: Document collection to search. One of: 'general',
            'technical', 'legal'. Defaults to 'general'.
        max_results: Maximum number of results to return (1–20).

    Returns:
        dict with:
            status: 'success' or 'error'.
            results: list of dicts, each with 'title', 'url', 'snippet'.
            total_found: int — total matching documents.
    """
    hits = vector_store.search(query, collection=collection, top_k=max_results)
    return {
        "status": "success",
        "results": [{"title": h.title, "url": h.url, "snippet": h.snippet} for h in hits],
        "total_found": hits.total,
    }

Pre-Built Toolbox Connectors

Rather than hand-rolling connectors for common systems, the toolbox extra ships maintained, versioned wrappers:

from google.adk.toolbox.github import GitHubToolset
from google.adk.toolbox.jira import JiraToolset
from google.adk.toolbox.mongodb import MongoDBToolset
from google.adk import Agent

github_tools  = GitHubToolset(repo="acme/backend", token=os.environ["GITHUB_TOKEN"])
jira_tools    = JiraToolset(project="ENG", base_url="https://acme.atlassian.net")
mongo_tools   = MongoDBToolset(connection_string=os.environ["MONGO_URI"], read_only=True)

devops_agent = Agent(
    name="devops_agent",
    model="gemini-3-flash",
    instruction="Triage failing CI runs: check GitHub Actions, file/update Jira tickets, and query build metrics from MongoDB.",
    tools=[*github_tools.get_tools(), *jira_tools.get_tools(), *mongo_tools.get_tools()],
)

MCP Tool Integration

Connect any Model Context Protocol server as a toolset — ADK discovers tools from the MCP server automatically. With Agent Registry now GA, MCP servers registered in your org can also be discovered by name instead of a hardcoded connection string.

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

async def build_agent_with_mcp():
    mcp_tools, exit_stack = await MCPToolset.from_server(
        connection_params=StdioServerParameters(
            command="npx",
            args=["-y", "@modelcontextprotocol/server-filesystem", "/workspace"],
        )
    )
    agent = Agent(
        name="fs_agent",
        model="gemini-3-flash",
        tools=mcp_tools,    # tools auto-discovered from MCP server
    )
    return agent, exit_stack

OpenAPI Tool Integration

from google.adk.tools.openapi_tool import OpenAPIToolset

async def build_api_agent():
    toolset = await OpenAPIToolset.from_spec(
        spec_url="https://api.example.com/openapi.json",
        operations_filter=lambda op: op.method == "GET",
    )
    agent = Agent(
        name="api_agent",
        model="gemini-3-flash",
        tools=toolset.tools,
    )
    return agent

Tool Best Practices

Skills for Agents

Agent Skills are reusable, self-contained capability packages that work within AI context-window limits. They differ from raw tools by encapsulating a higher-level behaviour with its own context, prompt fragments, and internal logic.

Skills vs Tools vs Sub-Agents vs Graph Nodes

Using a Pre-built Skill

from google.adk.skills import load_skill
from google.adk import Agent

# Load a pre-built skill from the Agent Skills registry
web_research_skill = load_skill("google/web-research")
summariser_skill   = load_skill("google/summarise-document")

agent = Agent(
    name="research_agent",
    model="gemini-3-flash",
    tools=[web_research_skill, summariser_skill],
)

Building a Custom Skill

A skill is a Python class that implements the BaseSkill interface, packaging related tools + context instructions together:

from google.adk.skills import BaseSkill
from dataclasses import dataclass

@dataclass
class CRMSkill(BaseSkill):
    """CRM operations skill — search contacts, create deals, update pipeline.

    Optimised to minimise context-window usage by pre-fetching commonly
    needed metadata in a single batch call rather than individual lookups.
    """
    crm_url: str
    api_key: str

    @property
    def name(self) -> str:
        return "crm_skill"

    @property
    def description(self) -> str:
        return (
            "Search CRM contacts, create or update deals, "
            "and manage sales pipeline stages."
        )

    def get_tools(self) -> list:
        return [self._search_contacts, self._create_deal, self._update_stage]

    def _search_contacts(self, name: str = "", email: str = "") -> dict:
        """Search CRM for contacts by name or email."""
        results = crm_api.search(name=name, email=email, url=self.crm_url)
        return {"status": "success", "contacts": results}

    def _create_deal(self, contact_id: str, title: str, value: float) -> dict:
        """Create a new deal in the CRM pipeline."""
        deal = crm_api.create_deal(contact_id=contact_id, title=title, value=value)
        return {"status": "success", "deal_id": deal.id}

    def _update_stage(self, deal_id: str, stage: str) -> dict:
        """Update the pipeline stage of an existing deal."""
        crm_api.update(deal_id=deal_id, stage=stage)
        return {"status": "success"}

# Use the skill in an agent
crm_skill = CRMSkill(crm_url="https://crm.example.com", api_key="...")
agent = Agent(
    name="sales_agent",
    model="gemini-3-flash",
    tools=crm_skill.get_tools(),
)

Context-Efficiency Tips

• Batch lookups: Skills should fetch all needed context in a single tool call rather than making multiple sequential calls.
• Return summaries: Return compact summaries of large data sets; include a fetch_details(id) tool for when the LLM needs more.
• Use state caching: Cache expensive lookups in session state so sibling tools in the same skill don't repeat the same API call.
• Skill composition: Compose coarse-grained skills from fine-grained tools rather than exposing every low-level function directly to the LLM.

Callbacks

Callbacks hook into the agent execution lifecycle without modifying core logic. They are the primary extension point for logging, guardrails, caching, and side effects.

from google.adk.agents.callback_context import CallbackContext
from google.adk.models import LlmRequest, LlmResponse
import time

# Example: LLM call logger + latency tracker
def before_model_log(ctx: CallbackContext, request: LlmRequest):
    ctx.state["_llm_start"] = time.time()
    print(f"[LLM] Calling {request.model} | tokens≈{len(str(request.contents))//4}")

def after_model_log(ctx: CallbackContext, response: LlmResponse):
    elapsed = time.time() - ctx.state.pop("_llm_start", time.time())
    print(f"[LLM] Done in {elapsed:.2f}s")

agent = Agent(
    name="monitored_agent",
    model="gemini-3-flash",
    before_model_callback=before_model_log,
    after_model_callback=after_model_log,
    ...
)

2.0 Migration Notes

ADK Python 2.0 reached GA on May 19, 2026; ADK Go 2.0 followed on June 30, 2026. Both introduce the Workflow Runtime as a genuinely additive layer — existing 1.x LlmAgent, SequentialAgent, ParallelAgent, and LoopAgent code keeps working — but a few things are worth reviewing before upgrading a production app.

Safe Upgrade Path

# If you're not ready to move to 2.0 yet, pin the compatible-release operator:
pip install "google-adk~=1.0"

# When ready, upgrade and re-run your eval suite before promoting to prod
pip install --upgrade google-adk
adk eval my_agent ./evals/regression_suite.json

Deployment

Cloud Run (Recommended for Most Teams)

# Build and push container
gcloud builds submit --tag gcr.io/PROJECT_ID/my-agent

# Deploy to Cloud Run
gcloud run deploy my-agent \
  --image gcr.io/PROJECT_ID/my-agent \
  --region us-central1 \
  --platform managed \
  --allow-unauthenticated \
  --set-env-vars GOOGLE_CLOUD_PROJECT=PROJECT_ID

Agent Runtime (Production Grade, formerly Agent Engine)

Agent Runtime provides session persistence, horizontal scaling, and built-in observability — with OpenTelemetry tracing enabled by default for new ADK agent deployments as of the 2026 release wave:

# Deploy using ADK CLI
adk deploy cloud_run \
  --project PROJECT_ID \
  --region us-central1 \
  --agent_module my_agent \
  --service_name my-agent-service \
  --verbosity debug     # detailed Cloud Run deploy logging

# Or use the agents-cli for the full managed Agent Runtime
agents deploy --source . --name my-agent --region us-central1

Deployment Checklist

• Set max_llm_calls in RunConfig and per-node timeout/RetryConfig to prevent runaway agents and stuck graphs.
• Use a production-grade session service (Cloud Spanner, Firestore, or Postgres) — verify your schema handles the 2.0 node_info/output event fields.
• Rely on default-on OpenTelemetry tracing on Agent Runtime, or export to your own OTel collector / MLflow for self-hosted observability.
• Store all secrets in Secret Manager — never in environment variables baked into container images.
• Set up rate limiting and budget alerts in Google Cloud Billing.
• Write evaluation tests with adk eval before promoting to production, and re-run them after any ADK version bump.
• If exposing agents for A2A delegation, register them in Agent Registry so other teams can discover them without hardcoded endpoints.

Reference Links

• Python Quickstart Installation, project setup, first agent
• ADK 2.0 Overview & Migration Guide Workflow Runtime, Task API, breaking changes
• LLM Agents LlmAgent / Agent reference — all parameters
• Graph Routes Workflow, edges, routers, JoinNode
• Classic Workflow Agents SequentialAgent, ParallelAgent, LoopAgent
• Multi-Agent Systems & A2A Agent teams, orchestration, A2A protocol
• Custom Agents Extending BaseAgent
• Agent Config GenerateContentConfig, RunConfig, RetryConfig reference
• Custom Tools Function tools, MCP tools, OpenAPI toolsets, Toolbox connectors
• Skills for Agents Pre-built skills, custom skill development
• Callbacks Lifecycle hooks, guardrails, logging patterns
• Deploy to Cloud Run Container deployment guide
• Agent Runtime on Gemini Enterprise Agent Platform Managed deployment, formerly Vertex AI Agent Engine
• Evaluation Testing agents with adk eval
• ADK Python GitHub Source code, changelog, issues, examples
• ADK Go GitHub Idiomatic Go implementation, GA since June 2026
• Python API Reference Full class / method reference incl. google.adk.workflow
• google-adk on PyPI Latest release, changelog, extras