Google Agents Cli Adk Code

ADK Code Reference

Before using this skill, activate /google-agents-cli-workflow first — it contains the required development phases and scaffolding steps.

Prerequisites

1. Run agents-cli info — if it shows project config, skip to the reference below 2. If no project exists: run agents-cli scaffold create <name> 3. If user has existing code: run agents-cli scaffold enhance .

Do NOT write agent code until a project is scaffolded.

Python only for now. This reference currently covers the Python ADK SDK.

Support for other languages is coming soon.

Quick Reference — Most Common Patterns

from google.adk.agents import Agent

def get_weather(city: str) -> dict:
    """Get current weather for a city."""
    return {"city": city, "temp": "22°C", "condition": "sunny"}

root_agent = Agent(
    name="my_agent",
    model="gemini-flash-latest",
    instruction="You are a helpful assistant that ...",
    tools=[get_weather],
)

---

References

The first two are cheatsheets for common patterns; for broad or deep knowledge, go to the source (docs index or installed package).

Related Skills

• /google-agents-cli-workflow — Development workflow, coding guidelines, and operational rules
• /google-agents-cli-scaffold — Project creation and enhancement with agents-cli scaffold create / scaffold enhance
• /google-agents-cli-eval — Evaluation methodology, dataset schema, and the eval-fix loop
• /google-agents-cli-deploy — Deployment targets, CI/CD pipelines, and production workflows

ADK Python Cheatsheet

1. Core Concepts & Project Structure

Essential Primitives

• `Agent`: The core intelligent unit. Can be LlmAgent (LLM-driven) or BaseAgent (custom/workflow).
• `Tool`: Callable function providing external capabilities (FunctionTool, AgentTool, etc.).
• `Session`: A stateful conversation thread with history (events) and short-term memory (state).
• `State`: Key-value dictionary within a Session for transient conversation data.
• `Runner`: The execution engine; orchestrates agent activity and event flow.
• `Event`: Atomic unit of communication; carries content and side-effect actions.

Standard Project Layout

your_project_root/
├── <agent_name>/ or app/    # Agent code directory
│   ├── __init__.py
│   ├── agent.py            # Contains root_agent definition
│   ├── tools.py            # Custom tool functions
│   └── .env                # Environment variables
├── tests/
│   ├── eval/
│   │   ├── eval_config.yaml    # Eval criteria and thresholds
│   │   └── datasets/           # Eval datasets (JSON)
│   ├── integration/
│   └── unit/
└── pyproject.toml or requirements.txt

---

2. Agent Definitions (LlmAgent)

Basic Setup

from google.adk.agents import Agent

def get_weather(city: str) -> dict:
    """Returns weather for a city."""
    return {"status": "success", "weather": "sunny", "temp": 72}

my_agent = Agent(
    name="weather_agent",
    model="gemini-flash-latest",
    instruction="You help users check the weather. Use the get_weather tool.",
    description="Provides weather information.",  # Important for multi-agent delegation
    tools=[get_weather]
)

Key Configuration Options

from google.genai import types as genai_types
from google.adk.agents import Agent

agent = Agent(
    name="my_agent",
    model="gemini-flash-latest",
    instruction="Your instructions here. Use {state_key} for dynamic injection.",
    description="Description for delegation.",

    # LLM generation parameters
    generate_content_config=genai_types.GenerateContentConfig(
        temperature=0.2,
        max_output_tokens=1024,
    ),

    # Save final output to state
    output_key="agent_response",

    # Control history sent to LLM
    include_contents='default',  # 'default' or 'none'

    # Delegation control
    disallow_transfer_to_parent=False,
    disallow_transfer_to_peers=False,

    # Sub-agents for delegation
    sub_agents=[specialist_agent],

    # Tools
    tools=[my_tool],

    # Callbacks
    before_agent_callback=my_callback,
    after_agent_callback=my_callback,
    before_model_callback=my_callback,
    after_model_callback=my_callback,
    before_tool_callback=my_callback,
    after_tool_callback=my_callback,
)

Structured Output with Pydantic

Warning: Using output_schema disables tool calling and delegation.

from pydantic import BaseModel, Field
from typing import Literal

class Evaluation(BaseModel):
    grade: Literal["pass", "fail"] = Field(description="The evaluation result.")
    comment: str = Field(description="Explanation of the grade.")

evaluator = Agent(
    name="evaluator",
    model="gemini-flash-latest",
    instruction="Evaluate the input and provide structured feedback.",
    output_schema=Evaluation,
    output_key="evaluation_result",
)

Instruction Best Practices

# Use dynamic state injection with {state_key} placeholders
instruction = """
You are a {role} assistant.
User preferences: {user_preferences}

Rules:
- Always use tools when available
- Never make up information
"""

---

3. Orchestration with Workflow Agents

Workflow agents provide deterministic control flow without LLM orchestration.

These are BaseAgent-family composites (SequentialAgent, ParallelAgent, LoopAgent). For the new graph-based Workflow API introduced in ADK 2.0, see references/adk-workflows.md.

SequentialAgent

Executes sub-agents in order. State changes propagate to subsequent agents.

from google.adk.agents import SequentialAgent, Agent

summarizer = Agent(
    name="summarizer",
    model="gemini-flash-latest",
    instruction="Summarize the input.",
    output_key="summary"
)

question_gen = Agent(
    name="question_generator",
    model="gemini-flash-latest",
    instruction="Generate questions based on: {summary}"
)

pipeline = SequentialAgent(
    name="pipeline",
    sub_agents=[summarizer, question_gen],
)

ParallelAgent

Executes sub-agents concurrently. Use distinct output_keys to avoid race conditions.

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

fetch_a = Agent(name="fetch_a", ..., output_key="data_a")
fetch_b = Agent(name="fetch_b", ..., output_key="data_b")

merger = Agent(
    name="merger",
    instruction="Combine data_a: {data_a} and data_b: {data_b}"
)

pipeline = SequentialAgent(
    name="full_pipeline",
    sub_agents=[
        ParallelAgent(name="fetchers", sub_agents=[fetch_a, fetch_b]),
        merger
    ]
)

LoopAgent

Repeats sub-agents until max_iterations or an event with escalate=True.

from google.adk.agents import LoopAgent

refinement_loop = LoopAgent(
    name="refinement_loop",
    sub_agents=[evaluator, refiner, escalation_checker],
    max_iterations=5,
)

For a production LoopAgent with EscalationChecker, BuiltInPlanner, and grounding citations, see /google-agents-cli-workflow Phase 1.

---

4. Multi-Agent Systems & Communication

Communication Methods

1. Shared State: Agents read/write session.state. Use output_key for convenience.

2. LLM Delegation: Agent transfers control to a sub-agent based on reasoning.

    coordinator = Agent(
        name="coordinator",
        instruction="Route to sales_agent for sales, support_agent for help.",
        sub_agents=[sales_agent, support_agent],
    )

3. AgentTool: Invoke another agent as a tool (parent stays in control).

    from google.adk.tools import AgentTool

    root = Agent(
        name="root",
        tools=[AgentTool(specialist_agent)],
    )

4. Task Delegation (ADK 2.0): Set mode on a sub-agent for structured, schema-typed delegation — the coordinator gets a request_task_{name} tool; the sub-agent returns typed output via the auto-injected finish_task tool.

    from pydantic import BaseModel

    class ResearchOutput(BaseModel):
        summary: str

    researcher = Agent(
        name="researcher",
        model="gemini-flash-latest",
        mode="task",                        # 'chat' (default) | 'task' | 'single_turn'
        output_schema=ResearchOutput,
        description="Researches a topic.",  # required for delegation
        instruction="Research the topic, then call finish_task.",
    )
    coordinator = Agent(name="coordinator", model="gemini-flash-latest", sub_agents=[researcher])

Modes: task (multi-turn, structured I/O) · single_turn (autonomous, no user turn). Sub-agents need a description; default I/O schemas (goal/background in, result out) are used if none set. Disabled inside graph Workflows.

---

5. Building Custom Agents (BaseAgent)

For custom orchestration logic beyond workflow agents.

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

class ConditionalRouter(BaseAgent):
    async def _run_async_impl(
        self, ctx: InvocationContext
    ) -> AsyncGenerator[Event, None]:
        # Read state
        user_type = ctx.session.state.get("user_type", "regular")

        # Custom routing logic
        if user_type == "premium":
            agent = self.premium_agent
        else:
            agent = self.regular_agent

        # Run selected agent
        async for event in agent.run_async(ctx):
            yield event

class EscalationChecker(BaseAgent):
    """Stops a LoopAgent when condition is met."""
    async def _run_async_impl(
        self, ctx: InvocationContext
    ) -> AsyncGenerator[Event, None]:
        result = ctx.session.state.get("evaluation")
        if result and result.get("grade") == "pass":
            yield Event(author=self.name, actions=EventActions(escalate=True))
        else:
            yield Event(author=self.name)

---

6. Models Configuration

Google Gemini (Default)

# AI Studio (dev)
# Set: GOOGLE_API_KEY, GOOGLE_GENAI_USE_VERTEXAI=False

# Vertex AI (prod)
# Set: GOOGLE_CLOUD_PROJECT, GOOGLE_CLOUD_LOCATION, GOOGLE_GENAI_USE_VERTEXAI=True

agent = Agent(model="gemini-flash-latest", ...)

Other Models via LiteLLM

from google.adk.models.lite_llm import LiteLlm

agent = Agent(model=LiteLlm(model="openai/gpt-4o"), ...)
agent = Agent(model=LiteLlm(model="anthropic/claude-sonnet-4-20250514"), ...)
agent = Agent(model=LiteLlm(model="ollama_chat/llama3:instruct"), ...)

Vertex AI Native Models

from google.adk.models import Gemini

# Vertex AI hosted Gemini (set GOOGLE_GENAI_USE_VERTEXAI=True)
agent = Agent(model=Gemini(model="gemini-flash-latest"), ...)

Provider guides: Anthropic, Ollama, vLLM, LiteLLM

---

7. Tools: The Agent's Capabilities

Function Tool Basics

from google.adk.tools import ToolContext

def search_database(
    query: str,
    limit: int,
    tool_context: ToolContext  # Optional, for state access
) -> dict:
    """Searches the database for records matching the query.

    Args:
        query: The search query string.
        limit: Maximum number of results to return.

    Returns:
        dict with 'status' and 'results' keys.
    """
    # Access state if needed
    user_id = tool_context.state.get("user_id")

    # Tool logic here
    results = db.search(query, limit=limit, user=user_id)

    return {"status": "success", "results": results}

Tool Rules:

• Use clear docstrings (sent to LLM)
• Type hints required, NO default values
• Return a dict (JSON-serializable)
• Don't mention tool_context in docstring

ToolContext Capabilities

async def my_tool(query: str, tool_context: ToolContext) -> dict:
    # Read/write state
    tool_context.state["key"] = "value"

    # Trigger escalation (stops LoopAgent)
    tool_context.actions.escalate = True

    # Artifacts — see Artifacts section below for full API
    await tool_context.save_artifact("file.txt", part)

    # Memory search
    results = await tool_context.search_memory("query")

    return {"status": "success"}

Built-in Tools

from google.adk.tools import google_search
from google.adk.tools import VertexAiSearchTool
from google.adk.tools.load_web_page import load_web_page
from google.adk.code_executors import BuiltInCodeExecutor

# Google Search grounding
agent = Agent(tools=[google_search], ...)

# Agent Platform Search grounding (your own data)
agent = Agent(tools=[VertexAiSearchTool(data_store_id="projects/P/locations/L/collections/default_collection/dataStores/DS")], ...)

# Web page loading
agent = Agent(tools=[load_web_page], ...)

# Code execution (model-internal)
agent = Agent(code_executor=BuiltInCodeExecutor(), ...)

# Managed sandbox (Vertex AI Code Interpreter) — see /google-agents-cli-workflow Phase 1
# from google.adk.code_executors import VertexAiCodeExecutor
# agent = Agent(code_executor=VertexAiCodeExecutor(optimize_data_file=True, stateful=True), ...)

`google_search` is model-internal grounding, not a regular tool. Mixing it with FunctionTools disables Automatic Function Calling (AFC) for all tools. If you need search alongside custom tools, consider a sub-agent architecture or a custom search function — see the deep-search sample for a working pattern. For eval implications, see the eval guide's builtin-tools-eval reference.

Tool Confirmation

from google.adk.tools import FunctionTool

# Simple confirmation
sensitive_tool = FunctionTool(delete_record, require_confirmation=True)

# Conditional confirmation
def needs_approval(amount: float, **kwargs) -> bool:
    return amount > 1000

transfer_tool = FunctionTool(transfer_money, require_confirmation=needs_approval)

Human-in-the-Loop (pause & resume)

Pause a run to ask the user something, then resume. This is a general runtime feature (not workflow-specific). Enable resumption at the app level:

from google.adk.apps import App, ResumabilityConfig

app = App(name="my_app", root_agent=root_agent,
          resumability_config=ResumabilityConfig(is_resumable=True))

• Let the model ask: add the built-in request_input tool (from google.adk.tools import request_input) to tools= — the model calls it when it needs clarification.
• Approval gate inside a tool: tool_context.request_confirmation(hint="Approve this transfer?"), or FunctionTool(fn, require_confirmation=...) (above).
• Custom long-running tool: wrap a function with LongRunningFunctionTool(fn) to pause until an external result arrives.

The user's reply is read from ctx.resume_inputs (available on ToolContext and CallbackContext). Inside graph workflows the same mechanism is node-based — see adk-workflows.md §7.

Tool Authentication

Helpers: from google.adk.tools.openapi_tool.auth.auth_helpers import token_to_scheme_credential, service_account_dict_to_scheme_credential. Pass auth_scheme + auth_credential to OpenAPIToolset(...). Full docs

OpenAPI Tools

from google.adk.tools.openapi_tool.openapi_spec_parser.openapi_toolset import OpenAPIToolset

toolset = OpenAPIToolset(spec_str=open("openapi.json").read(), spec_str_type="json")
agent = Agent(name="api_agent", tools=[toolset], ...)

Pass auth_scheme + auth_credential from the auth helpers above for authenticated APIs. Tool names derive from operationId (snake_case, max 60 chars). Full docs

MCP Tools

Connect to MCP servers to use external tools. Use StdioConnectionParams for local dev, SseConnectionParams for production.

from google.adk.tools.mcp_tool import McpToolset
from google.adk.tools.mcp_tool.mcp_session_manager import StdioConnectionParams, SseConnectionParams
from mcp import StdioServerParameters

# Local MCP server via stdio
agent = Agent(
    name="my_agent",
    tools=[
        McpToolset(
            connection_params=StdioConnectionParams(
                server_params=StdioServerParameters(
                    command="npx",
                    args=["-y", "@modelcontextprotocol/server-filesystem", "/absolute/path"],
                ),
            ),
            tool_filter=["list_directory", "read_file"],  # optional: restrict exposed tools
        )
    ],
    ...
)

# Remote MCP server via SSE (production)
McpToolset(
    connection_params=SseConnectionParams(url="https://mcp.example.com/sse"),
)

Gotchas:

• Paths must be absolute, not relative.
• Agent definition must be synchronous (not async) for deployment.
• Node.js/npx required for npm-based MCP servers — add to Dockerfile if containerizing.

---

8. Context, State, and Memory

State Prefixes

# Session-specific (default)
state["booking_step"] = 2

# User-persistent (across sessions)
state["user:preferred_language"] = "en"

# App-wide (all users)
state["app:total_queries"] = 1000

# Temporary (current invocation only)
state["temp:intermediate_result"] = data

Session Service Options

from google.adk.sessions import InMemorySessionService
# For dev: InMemorySessionService()
# For prod: VertexAiSessionService(), DatabaseSessionService()

Session Rewind

Roll back a session to the state before a specific invocation (useful for debugging or user-initiated undo):

from google.adk.runners import InMemoryRunner

runner = InMemoryRunner(agent=root_agent, app_name="my_app")

# Rewind to state before a given invocation
await runner.rewind_async(
    user_id=user_id,
    session_id=session.id,
    rewind_before_invocation_id=invocation_id,  # exclusive: state before this call
)

Note: Restores session-level state and artifacts only; app/user-scoped state is unaffected.

Artifacts (File Storage)

Store and retrieve binary data (PDFs, images, audio) scoped to session or user:

from google.adk.artifacts import InMemoryArtifactService, GcsArtifactService
from google.genai import types

# Configure runner with artifact service
runner = Runner(
    agent=root_agent,
    app_name="app",
    session_service=session_service,
    artifact_service=InMemoryArtifactService(),  # or GcsArtifactService(bucket_name="my-bucket")
)

# In a tool or callback:
async def save_file(data: bytes, tool_context: ToolContext) -> dict:
    part = types.Part(inline_data=types.Blob(mime_type="application/pdf", data=data))
    version = await tool_context.save_artifact("report.pdf", part)       # session-scoped
    await tool_context.save_artifact("user:profile.png", part)           # user-scoped
    artifact = await tool_context.load_artifact("report.pdf")            # latest version
    artifact_v0 = await tool_context.load_artifact("report.pdf", version=0)
    names = await tool_context.list_artifacts()
    return {"status": "saved", "version": version}

Namespace prefixes: plain name = session-scoped · "user:" = persistent across sessions

Memory (Long-term Knowledge)

InMemoryMemoryService (Dev)

In-memory implementation for local development. Memories don't persist across restarts.

from google.adk.memory import InMemoryMemoryService

memory_service = InMemoryMemoryService()
# Add session to memory after conversation
await memory_service.add_session_to_memory(session)
# Search later
results = await memory_service.search_memory(app_name=app_name, user_id=user_id, query="query")

Memory Bank (Long-term Memory)

Managed cross-session memory that persists user preferences, remembers facts across sessions, and learns from conversations over time. See the `memory-bank` sample for a complete implementation.

from google.adk.agents.callback_context import CallbackContext
from google.adk.tools.preload_memory_tool import PreloadMemoryTool

# PreloadMemoryTool retrieves memories at the start of each turn and injects
# them into the system instruction. Alternative: LoadMemoryTool() — the model
# calls it on-demand when it decides memories are needed.
root_agent = Agent(
    ...,
    tools=[PreloadMemoryTool()],
    after_agent_callback=generate_memories_callback,
)

# Alternative: callback_context.add_events_to_memory(events=...) to send only
# a subset of events, which is better for incremental processing.
async def generate_memories_callback(callback_context: CallbackContext):
    """Sends the session's events to Memory Bank for memory generation."""
    await callback_context.add_session_to_memory()
    return None

Context Caching

Cache large context windows (system prompt + docs) to reduce latency and cost. Transparent to agent code.

from google.adk.apps import App
from google.adk.agents.context_cache_config import ContextCacheConfig

app = App(
    name="my_app",
    root_agent=root_agent,
    context_cache_config=ContextCacheConfig(
        min_tokens=2048,     # only cache if context exceeds this
        ttl_seconds=1800,    # cache lifetime (default 1800)
        cache_intervals=10,  # re-cache every N invocations
    ),
)

Context Compaction

Prevent context overflow on long sessions by summarizing older events in a sliding window:

from google.adk.apps import App
from google.adk.apps.app import EventsCompactionConfig
from google.adk.apps.llm_event_summarizer import LlmEventSummarizer
from google.adk.models import Gemini

app = App(
    name="my_app",
    root_agent=root_agent,
    events_compaction_config=EventsCompactionConfig(
        compaction_interval=20,  # summarize every 20 events
        overlap_size=3,          # include last 3 events in next window for continuity
        # Optional: custom summarizer model
        summarizer=LlmEventSummarizer(llm=Gemini(model="gemini-flash-latest")),
    ),
)

App Name

The App(name=...) parameter must match the agent directory name (default: app). A mismatch causes "Session not found" errors during evaluation because the runner infers the app name from the directory path.

# CORRECT — matches the "app" directory
app = App(name="app", root_agent=root_agent)

# WRONG — causes eval failures
app = App(name="my_custom_agent", root_agent=root_agent)

---

9. Callbacks

Callback Types

from google.adk.agents.callback_context import CallbackContext
from google.adk.models.llm_request import LlmRequest
from google.adk.models.llm_response import LlmResponse
from google.adk.tools import BaseTool, ToolContext
from google.genai import types as genai_types

# Callbacks are invoked by keyword — parameter names must match exactly.

# Agent lifecycle
async def before_agent_callback(callback_context: CallbackContext) -> None:
    callback_context.state["started"] = True

async def after_agent_callback(callback_context: CallbackContext) -> genai_types.Content | None:
    # Return None to continue, or Content to override
    return None

# Model interaction
async def before_model_callback(callback_context: CallbackContext, llm_request: LlmRequest) -> LlmResponse | None:
    # Return None to continue, or LlmResponse to skip model call
    return None

async def after_model_callback(callback_context: CallbackContext, llm_response: LlmResponse) -> LlmResponse | None:
    # Return None to continue, or modified LlmResponse
    return None

# Tool execution
async def before_tool_callback(tool: BaseTool, args: dict, tool_context: ToolContext) -> dict | None:
    # Return None to continue, or dict to skip tool and use as result
    return None

async def after_tool_callback(tool: BaseTool, args: dict, tool_context: ToolContext, tool_response: dict) -> dict | None:
    # Return None to continue, or modified dict
    return None

Common Pattern

# Initialize state before agent runs
async def init_state(callback_context: CallbackContext) -> None:
    if "preferences" not in callback_context.state:
        callback_context.state["preferences"] = {}

agent = Agent(before_agent_callback=init_state, ...)

---

10. Plugins

Global callback hooks across all agents/tools/LLMs. Use for cross-cutting concerns (logging, guardrails); use callbacks for per-agent logic.

from google.adk.plugins.base_plugin import BasePlugin
from google.adk.apps import App

class MyPlugin(BasePlugin):
    async def before_model_callback(self, *, callback_context, llm_request):
        return None  # return None to observe, return value to intervene

# Register via App — plugins run BEFORE agent-level callbacks
app = App(name="my_app", root_agent=root_agent, plugins=[MyPlugin()])
runner = Runner(app=app, session_service=...)

Built-in plugins: ReflectAndRetryToolPlugin (retry failed tools), BigQueryAgentAnalyticsPlugin (log to BQ), ContextFilterPlugin (reduce context size), GlobalInstructionPlugin (shared system prompt), SaveFilesAsArtifactsPlugin, LoggingPlugin, DebugLoggingPlugin, MultimodalToolResultsPlugin.

Hooks: before/after_agent_callback, before/after_model_callback, before/after_tool_callback, on_model_error_callback, on_tool_error_callback, on_user_message_callback, before/after_run_callback, on_event_callback. Full docs

Safety Guardrails

Use before_model_callback to filter input or after_model_callback to filter output. Return None to pass through, or return a modified LlmResponse to block/replace. Evaluate with the safety metric. Full docs

---

11. A2A Protocol

Requires pip install google-adk[a2a].

# Expose an agent as an A2A service
# Prefer scaffolding over manual code — use --agent adk_a2a (see /google-agents-cli-scaffold)
from google.adk.a2a.utils.agent_to_a2a import to_a2a
from a2a.types import AgentCard
to_a2a(root_agent, port=8001)

# Consume a remote A2A agent
from google.adk.agents.remote_a2a_agent import RemoteA2aAgent, AGENT_CARD_WELL_KNOWN_PATH
remote = RemoteA2aAgent(
    name="remote_agent",
    description="...",
    agent_card=f"http://remote-host:8001{AGENT_CARD_WELL_KNOWN_PATH}",
)

---

12. Event-Driven / Ambient Agents

Ambient agents process events (Pub/Sub, Eventarc, schedules) autonomously. ADK provides built-in trigger endpoints that handle payload decoding, session creation, concurrency, and retries.

Deployment: Trigger endpoints require Cloud Run or GKE. Agent Runtime does not support event-driven or scheduled triggers.

from google.adk.cli.fast_api import get_fast_api_app

app = get_fast_api_app(
    agents_dir=AGENTS_DIR,
    web=False,
    trigger_sources=["pubsub", "eventarc"],  # enables /apps/{app}/trigger/pubsub and /trigger/eventarc
)

# CLI equivalent for local dev
adk api_server --trigger_sources "pubsub,eventarc" path/to/your/agent

Trigger endpoints handle: base64 decoding, CloudEvent parsing, per-event session creation (UUID), concurrency semaphore, and exponential backoff on transient errors.

Sessions are ephemeral by default (InMemorySessionService); use DatabaseSessionService for audit trails. Pub/Sub and Eventarc have a 10-minute processing limit. For non-GCP sources, use adk api_server --auto_create_session with the /run endpoint instead.

Scheduled / cron execution: Use Cloud Scheduler to publish to a Pub/Sub topic on a cron schedule, then connect the topic to the agent's /apps/{app}/trigger/pubsub endpoint. This is how you implement "run daily at 8 PM" — no custom scheduling code needed.

Since ambient agents have no interactive user, route outputs via structured logging (JSON stdout → Cloud Logging → Cloud Monitoring alerts), Pub/Sub, or tool-based integrations (email, Jira, Slack).

Before implementing an ambient agent, clone and study the production sample — it covers trigger wiring, middleware, structured logging, and Terraform. See the Notable Samples table in /google-agents-cli-workflow Phase 1. Full docs.

---

Quick Reference

Running Agents Programmatically

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

session_service = InMemorySessionService()
await session_service.create_session(app_name="app", user_id="user", session_id="s1")
runner = Runner(agent=my_agent, app_name="app", session_service=session_service)

async for event in runner.run_async(
    user_id="user", session_id="s1",
    new_message=types.Content(role="user", parts=[types.Part.from_text(text="Hello!")]),
):
    if event.is_final_response():
        print(event.content.parts[0].text)

ADK Built-in Tool Imports (Precision Required)

# CORRECT - imports the tool instance
from google.adk.tools.load_web_page import load_web_page

# WRONG - imports the module, not the tool
from google.adk.tools import load_web_page

Pass the imported tool directly to tools=[load_web_page], not tools=[load_web_page.load_web_page].

Factory Functions for Sub-agents

Use factory functions (not module-level instances) to avoid "agent already has a parent" errors. Always call the factory — passing the function reference fails with ValidationError: Input should be a valid dictionary or instance of BaseAgent.

def create_researcher():
    return Agent(name="researcher", ...)

root_agent = SequentialAgent(
    sub_agents=[create_researcher(), create_analyst()],  # call the functions!
    ...
)

Data flows between sequential sub-agents via conversation history and output_key state.

Further Reading

• ADK Documentation
• ADK Samples
• /google-agents-cli-workflow Phase 1 — curated production patterns (VertexAiCodeExecutor, BuiltInPlanner, grounding metadata, rate limiting, state capture)

---

Inspecting ADK Source Code

When you need to look up ADK internals, inspect the installed package directly:

# Find the ADK package location (use "uv run python" if using uv)
python -c "import google.adk; print(google.adk.__path__[0])"

ADK Package Directory Map

google/adk/
├── agents/           # Agent types (LlmAgent, BaseAgent, SequentialAgent, etc.)
├── tools/            # Tool implementations (FunctionTool, google_search, etc.)
├── sessions/         # Session services (InMemory, Database, VertexAI)
├── memory/           # Memory services
├── runners.py        # Runner and execution engine
├── events/           # Event types and actions
├── models/           # Model integrations (Gemini, LiteLLM, etc.)
├── code_executors/   # Code execution (BuiltInCodeExecutor, etc.)
├── evaluation/       # Eval framework (criteria, evaluators, etc.)
├── cli/              # ADK CLI internals (used by agents-cli playground, eval, etc.)
├── flows/            # LLM flow implementations
├── artifacts/        # Artifact services
└── auth/             # Authentication helpers

Use Glob/Grep/Read on the installed package to find exact implementations, method signatures, and configuration options.

For the full ADK documentation index, use curl https://adk.dev/llms.txt.

ADK Workflow API Cheatsheet

Requires google-adk >= 2.0.0. Python only.

Requires Python >= 3.11. The Workflow class itself does not support Live Streaming (Runner.run_live) — the graph engine needs strict control over event emission. Use a plain Agent for live/bidi flows. ADK 2.0 itself still ships Runner.run_live and LiveRequestQueue.

Official docs: Workflows overview · Graph routes · Collaboration · Data handling · Dynamic workflows · Human-in-the-loop

1. Core Concepts

A Workflow is a graph-based agent: nodes do work, edges define flow, START is the entry point.

from google.adk.workflow import Workflow

def greet(node_input: str) -> str:
  return f"Hello, {node_input}!"

root_agent = Workflow(
    name="greeter",
    edges=[('START', greet)],
)

Three building blocks: Nodes (functions, LLM agents, tools), Edges (connections with optional route conditions), START (built-in entry receiving user input).

Workflow Constructor

root_agent = Workflow(
    name="my_workflow",
    edges=[...],                # Edge definitions (or use graph= instead)
    description="",             # Agent description
    input_schema=None,          # Pydantic model for input validation
    output_schema=None,         # Pydantic model for the workflow's output
    state_schema=None,          # Pydantic model for state validation
    rerun_on_resume=True,       # Rerun workflow on resume (default: True)
    max_concurrency=None,       # Limit parallel node execution (None = no limit)
    retry_config=None,          # Default RetryConfig applied to nodes
    timeout=None,               # Whole-workflow timeout in seconds
    wait_for_output=False,      # Wait for dynamically scheduled child output
)

---

2. Node Types

Any "NodeLike" is accepted in edges and auto-wrapped:

Auto-wrapping is the recommended approach. Place functions, agents, and tools directly in edges — the framework wraps them automatically. You do not need to import or use internal wrapper classes directly.

---

3. Function Nodes

Most common node type. Parameter resolution:

from google.adk.agents.context import Context

def process(ctx: Context, node_input: Any, user_name: str) -> str:
  # node_input = predecessor output; user_name = ctx.state['user_name']
  # START outputs types.Content (not str) unless input_schema is set
  return f"{user_name}: {node_input}"

Return Types

• Value -> wrapped in Event(output=value), triggers downstream
• `None` -> no event emitted, no downstream trigger
• `Event` -> used directly (for routing or state updates)
• Generator -> yield multiple events; only the last with output triggers downstream

from google.adk.events.event import Event

def classify(node_input: str):
  if "urgent" in node_input:
    return Event(output=node_input, route="urgent")
  return Event(output=node_input, route="normal", state={"processed": True})

Auto Type Conversion

FunctionNode auto-converts dict inputs to Pydantic models based on type hints. Works for list[Model] and dict[str, Model] too.

node_input Type by Predecessor

@node Decorator & Explicit FunctionNode

from google.adk.workflow import node, FunctionNode, RetryConfig

@node
def my_func(node_input: str) -> str:
  return node_input

@node(name="custom", rerun_on_resume=True)
async def my_async(node_input: str) -> str:
  return node_input

# Explicit FunctionNode for full control (func is keyword-only)
fn = FunctionNode(
    func=my_func,
    retry_config=RetryConfig(max_attempts=3),
    timeout=30.0,                          # Seconds before timeout
    parameter_binding='state',             # 'state' (default) or 'node_input'
    auth_config=None,                      # Requires rerun_on_resume=True
    state_schema=None,                     # Pydantic model for state validation
)

---

4. Edge Patterns

# Sequential chain
edges = [('START', a), (a, b), (b, c)]

# Conditional routing (node returns Event with route=)
edges = [
    ('START', classifier),
    (classifier, success_handler, "success"),
    (classifier, error_handler, "error"),
    (classifier, fallback_handler, '__DEFAULT__'),  # Fallback route
]

# Fan-out (parallel branches)
edges = [('START', (branch_a, branch_b, branch_c))]

# Fan-in with JoinNode
from google.adk.workflow import JoinNode
join = JoinNode(name="merge")
edges = [((branch_a, branch_b), join), (join, final)]
# JoinNode output: {"branch_a": output_a, "branch_b": output_b}

# Looping (must have at least one routed edge — unconditional cycles rejected)
edges = [
    ('START', process),
    (process, check),
    (check, process, "continue"),
    (check, finish, "exit"),
]

Route values: str, bool, int. Multi-route fan-out: return Event(output=x, route=["a", "b"]). Edge matching multiple routes: (node, target, ["route_x", "route_y"]).

---

5. LLM Agent Nodes

Use google.adk.agents.LlmAgent in workflow edges — auto-wrapped internally, emits Event(output=...) for downstream data passing.

from google.adk.agents import LlmAgent
from pydantic import BaseModel

class DraftOutput(BaseModel):
  title: str
  content: str

writer = LlmAgent(
    name="writer",
    model="gemini-flash-latest",
    instruction="Write a draft based on the user's request.",
    output_schema=DraftOutput,  # Always set for structured output
    output_key="draft",         # Also store in state['draft']
)

agent = Workflow(
    name="pipeline",
    edges=[('START', writer), (writer, process_draft)],
)

Always use `output_schema` (Pydantic model) on LLM agents in workflows. Without it, output is types.Content which may cause type errors in downstream function nodes or serialization failures with JoinNode/database sessions.

---

6. Parallel Processing

ParallelWorker — process list items concurrently

from google.adk.workflow import node

@node(parallel_worker=True)
def process_item(node_input: int) -> int:
  return node_input * 2

# Input: [1, 2, 3] -> Output: [2, 4, 6]
agent = Workflow(
    name="parallel",
    edges=[('START', split_input), (split_input, process_item), (process_item, collect)],
)

Workers named {parent_name}@{index}. Input must be a list. Output is a list in same order.

Fan-Out / Fan-In — diamond pattern

from google.adk.workflow import JoinNode
join = JoinNode(name="merge")
edges = [
    ('START', splitter),
    (splitter, (branch_a, branch_b)),
    ((branch_a, branch_b), join),
    (join, combiner),  # combiner receives {"branch_a": ..., "branch_b": ...}
]

---

7. Human-in-the-Loop (HITL)

HITL is a general ADK feature (app-level ResumabilityConfig, the request_input tool, resume_inputs) — see adk-python.md "Human-in-the-Loop". Inside a workflow it works per node: a node yields RequestInput and reads replies from ctx.resume_inputs (keyed by interrupt_id).

from google.adk.events.request_input import RequestInput

async def multi_step(ctx: Context, node_input: str):
  if not ctx.resume_inputs:
    yield RequestInput(interrupt_id="ask_name", message="Name?")
    return
  if "ask_email" not in ctx.resume_inputs:
    yield RequestInput(interrupt_id="ask_email", message="Email?")
    return
  yield Event(output={"name": ctx.resume_inputs["ask_name"],
                      "email": ctx.resume_inputs["ask_email"]})

Node resume behavior: rerun_on_resume=False (default FunctionNode) → the user's response becomes the node output; rerun_on_resume=True (default LlmAgent) → the node reruns with ctx.resume_inputs populated. In loops: use a unique interrupt_id per iteration (e.g. f'review_{count}') to avoid infinite restarts.

---

8. State & Events

Context Properties

from google.adk.agents.context import Context

def my_node(ctx: Context, node_input: str) -> str:
  ctx.state.get("key", "default")   # Read state
  ctx.session.id                     # Session ID
  ctx.node_path                      # "Workflow/node_name"
  ctx.node                           # Current node
  ctx.run_id                         # Current execution ID
  ctx.attempt_count                  # 1 on first attempt (1-based)
  ctx.resume_inputs                  # HITL resume data (dict keyed by interrupt_id)
  ctx.interrupt_ids                  # Active interrupt IDs
  ctx.output                         # Node's result value (settable)
  ctx.route                          # Routing value (settable)
  return "result"

Dynamic Node Scheduling

async def orchestrator(ctx: Context, node_input: list) -> list:
  results = []
  for i, item in enumerate(node_input):
    result = await ctx.run_node(process_item, node_input=item)
    results.append(result)
  return results

ctx.run_node() requires rerun_on_resume=True on the calling node. Use use_as_output=True to delegate the node's output to the dynamic child.

Event Fields

State: Prefer Event over ctx.state

# Preferred — persisted in event history, replayable
def save(node_input: str):
  return Event(output=node_input, state={"key": node_input})

# Avoid — side effect, may be lost on replay
def save(ctx: Context, node_input: str) -> str:
  ctx.state["key"] = node_input
  return node_input

Data Serialization Rules

• Event.output must be JSON-serializable. BaseModel returns auto-converted via model_dump().
• output_key stores dicts (not BaseModel instances) — validate_schema() -> model_dump().
• ctx.state.get(key) returns a dict. Use MyModel(**data) to reconstruct typed access.

---

9. Retry Configuration

from google.adk.workflow import FunctionNode, RetryConfig

node = FunctionNode(
    func=flaky_call,           # func is keyword-only
    retry_config=RetryConfig(
        max_attempts=5,        # Default: None (treated as 5); 1 = no retry
        initial_delay=1.0,     # Seconds before first retry
        max_delay=60.0,        # Max seconds between retries
        backoff_factor=2.0,    # Delay multiplier per attempt
        jitter=1.0,            # Randomness factor (0.0 = none)
        exceptions=None,       # Exception types to retry (None = all)
    ),
)

Delay formula: min(initial_delay * backoff_factor^attempt, max_delay) * (1 + random(0, jitter))

---

10. Testing

Note: The testing utilities below (testing_utils, InMemoryRunner) are internal to the ADK repository. They are not part of the public google-adk package. For your own tests, use App + InMemoryRunner from google.adk.runners or write a custom test harness.

import pytest
from google.adk.workflow import Workflow
from google.adk.apps import App
from google.adk.runners import InMemoryRunner
from google.genai import types

@pytest.mark.asyncio
async def test_workflow():
  def step(node_input: str) -> str:
    return "done"

  agent = Workflow(name="test", edges=[('START', step)])
  app = App(name="test_app", root_agent=agent)
  runner = InMemoryRunner(app=app)

  session = await runner.session_service.create_session(
      app_name="test_app", user_id="test_user"
  )
  async for event in runner.run_async(
      user_id="test_user",
      session_id=session.id,
      new_message=types.Content(role="user", parts=[types.Part.from_text(text="hello")]),
  ):
    if event.output is not None:
      assert event.output == "done"

---

11. Import Paths

Workflow Core

Workflow Nodes (auto-wrapped)

Nodes are auto-wrapped when placed in edges. You do not need to import wrapper classes.

Events & Context

LLM Agent

App & Resumability

---

12. Best Practices

Use Pydantic Models, Not Raw Dicts

Always define BaseModel classes for node I/O, LLM output_schema, and structured data:

# Wrong: raw dicts
def lookup(node_input: dict[str, Any]) -> dict[str, Any]:
  return {"cost": 500}

# Correct: typed schemas
class FlightInfo(BaseModel):
  cost: int
  details: str

def lookup(node_input: Itinerary) -> FlightInfo:
  return FlightInfo(cost=500, details="Economy")

Emit Content Events for Web UI

event.output is internal — only event.content renders in the ADK web UI:

from google.genai import types

def final_output(node_input: str):
  yield Event(content=types.Content(role='model', parts=[types.Part.from_text(text=node_input)]))
  yield Event(output=node_input)

LLM agents emit content events automatically. Add them explicitly for function nodes with user-facing results.

Agent Directory Convention

my_workflow/
  __init__.py    # from . import agent
  agent.py       # root_agent = Workflow(...)

Advanced Patterns

• Nested workflows: A Workflow can be used as a node in another workflow
• Dynamic node scheduling: Use await ctx.run_node(func, node_input=item) at runtime (requires rerun_on_resume=True)
• Custom Node subclass: Subclass Node, implement run_node_impl(*, ctx, node_input) -> AsyncGenerator. Supports parallel_worker=True flag.
• Custom BaseNode: Subclass BaseNode, implement _run_impl(*, ctx, node_input) -> AsyncGenerator

Graph Validation Rules

1. START must exist and have no incoming edges 2. All non-START nodes must be reachable 3. No duplicate node names or edges 4. At most one __DEFAULT__ route per node 5. No unconditional cycles (cycles need at least one routed edge)

---

Further Reading

• Workflows overview
• Graph routes & conditional edges
• Agent collaboration & task mode
• Data handling & state
• Dynamic workflows
• Human-in-the-loop
• App class (workflow container)