Google ADK: Building Multi-Agent Systems with Agent Development Kit

# Create virtual environment
python -m venv .venv
source .venv/bin/activate  # macOS/Linux
# Windows: .venv\Scripts\activate

# Install ADK
pip install google-adk

mkdir my-adk-agent && cd my-adk-agent
npm init -y

# Install ADK packages
npm install @google/adk @google/adk-devtools
npm install -D typescript @types/node

# Option 1: Google AI Studio (simpler, for development)
GOOGLE_GENAI_USE_VERTEXAI=FALSE
GOOGLE_API_KEY=your_api_key_here

# Option 2: Vertex AI (for production)
GOOGLE_GENAI_USE_VERTEXAI=TRUE
GOOGLE_CLOUD_PROJECT=your-project-id
GOOGLE_CLOUD_LOCATION=us-central1

ADK Architecture
├── Agents              # Execution units (LLM-powered or workflow)
│   ├── LlmAgent        # Uses LLM for reasoning and decisions
│   ├── SequentialAgent # Executes sub-agents in order
│   ├── ParallelAgent   # Runs sub-agents concurrently
│   └── LoopAgent       # Repeats until condition met
├── Tools               # Capabilities agents can invoke
│   ├── FunctionTool    # Custom Python/TS functions
│   ├── AgentTool       # Other agents as callable tools
│   └── Built-in        # Google Search, Code Execution, etc.
├── Sessions            # Conversation state management
│   ├── State           # Current conversation data
│   └── Memory          # Cross-session knowledge
└── Runner              # Orchestration engine

my_agent/
├── __init__.py         # Makes it a Python package
├── agent.py            # Agent definition (must export root_agent)
└── .env                # Environment variables

# my_agent/agent.py
from google.adk.agents import Agent

root_agent = Agent(
    name="helpful_assistant",
    model="gemini-2.0-flash",
    description="A helpful assistant that answers questions.",
    instruction="""You are a helpful assistant.

    When users ask questions:
    1. Think through the problem step by step
    2. Provide clear, concise answers
    3. Ask for clarification if the question is ambiguous
    """
)

# Interactive web UI (best for development)
adk web

# Terminal chat interface
adk run my_agent

# API server (for integration)
adk api_server

from google.adk.agents import Agent

def get_weather(city: str, unit: str = "celsius") -> dict:
    """
    Get the current weather for a city.

    Args:
        city: The name of the city (e.g., "New York", "London")
        unit: Temperature unit - "celsius" or "fahrenheit" (default: celsius)

    Returns:
        dict with status, temperature, and conditions
    """
    # In production, call a real weather API
    weather_data = {
        "new york": {"temp": 22, "conditions": "Sunny"},
        "london": {"temp": 15, "conditions": "Cloudy"},
        "tokyo": {"temp": 28, "conditions": "Humid"},
    }

    city_lower = city.lower()
    if city_lower in weather_data:
        data = weather_data[city_lower]
        temp = data["temp"] if unit == "celsius" else (data["temp"] * 9/5) + 32
        return {
            "status": "success",
            "city": city,
            "temperature": temp,
            "unit": unit,
            "conditions": data["conditions"]
        }

    return {
        "status": "error",
        "error_message": f"Weather data not available for {city}"
    }


def get_current_time(timezone: str = "UTC") -> dict:
    """
    Get the current time in a specific timezone.

    Args:
        timezone: IANA timezone name (e.g., "America/New_York", "Europe/London")

    Returns:
        dict with status and current time
    """
    from datetime import datetime
    import pytz

    try:
        tz = pytz.timezone(timezone)
        current_time = datetime.now(tz)
        return {
            "status": "success",
            "timezone": timezone,
            "time": current_time.strftime("%Y-%m-%d %H:%M:%S"),
            "day_of_week": current_time.strftime("%A")
        }
    except Exception as e:
        return {
            "status": "error",
            "error_message": f"Invalid timezone: {timezone}"
        }


# Create agent with tools
root_agent = Agent(
    name="weather_time_agent",
    model="gemini-2.0-flash",
    description="Provides weather and time information",
    instruction="""You are a helpful assistant that provides weather and time information.

    When users ask about weather:
    - Use the get_weather tool to fetch current conditions
    - Always mention both temperature and conditions
    - Offer to convert units if the user might prefer different units

    When users ask about time:
    - Use the get_current_time tool with the appropriate timezone
    - If the user doesn't specify a timezone, ask for clarification
    - Include the day of the week in your response
    """,
    tools=[get_weather, get_current_time]
)

import { FunctionTool, LlmAgent } from '@google/adk';
import { z } from 'zod';

const getWeather = new FunctionTool({
  name: 'get_weather',
  description: 'Get the current weather for a city',
  parameters: z.object({
    city: z.string().describe('The city name'),
    unit: z.enum(['celsius', 'fahrenheit']).default('celsius')
  }),
  execute: async ({ city, unit }) => {
    // Implementation here
    return {
      status: 'success',
      city,
      temperature: 22,
      unit,
      conditions: 'Sunny'
    };
  }
});

export const rootAgent = new LlmAgent({
  name: 'weather_agent',
  model: 'gemini-2.0-flash',
  description: 'Provides weather information',
  instruction: 'You help users with weather queries.',
  tools: [getWeather]
});

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

# Agent with Google Search
search_agent = Agent(
    name="research_agent",
    model="gemini-2.0-flash",
    instruction="Research topics using Google Search.",
    tools=[google_search]
)

# Agent with code execution
code_agent = Agent(
    name="code_agent",
    model="gemini-2.0-flash",
    instruction="Write and execute Python code to solve problems.",
    tools=[code_execution]
)

from google.adk.agents import Agent

# Specialized subagents
billing_agent = Agent(
    name="billing_agent",
    model="gemini-2.0-flash",
    description="Handles billing questions, invoices, refunds, and payment issues.",
    instruction="""You are a billing specialist.

    You can help with:
    - Explaining charges and invoices
    - Processing refund requests
    - Updating payment methods
    - Resolving billing disputes

    Always verify the customer's identity before discussing account details.
    For refunds over $100, escalate to a human supervisor.
    """
)

technical_support_agent = Agent(
    name="technical_support",
    model="gemini-2.0-flash",
    description="Handles technical issues, bugs, troubleshooting, and product questions.",
    instruction="""You are a technical support specialist.

    You can help with:
    - Troubleshooting product issues
    - Explaining features and how to use them
    - Reporting bugs to the engineering team
    - Providing workarounds for known issues

    Always ask for error messages and steps to reproduce before diagnosing.
    """
)

general_info_agent = Agent(
    name="general_info",
    model="gemini-2.0-flash",
    description="Answers general questions about products, policies, and company information.",
    instruction="""You answer general questions about our company and products.

    You can help with:
    - Product information and comparisons
    - Company policies (returns, shipping, etc.)
    - Store locations and hours
    - General FAQs
    """
)

# Coordinator agent that routes to specialists
root_agent = Agent(
    name="customer_service_coordinator",
    model="gemini-2.0-flash",
    description="Main customer service agent that routes requests to specialists.",
    instruction="""You are the front-line customer service coordinator.

    Your job is to:
    1. Greet customers warmly
    2. Understand their needs
    3. Route them to the appropriate specialist:
       - billing_agent: For payment, invoice, or refund questions
       - technical_support: For product issues or troubleshooting
       - general_info: For general questions about products or policies

    If you're unsure which specialist to use, ask clarifying questions first.
    Always ensure a smooth handoff with context about what the customer needs.
    """,
    sub_agents=[billing_agent, technical_support_agent, general_info_agent]
)

from google.adk.agents import Agent, SequentialAgent

# Step 1: Fetch data
data_fetcher = Agent(
    name="data_fetcher",
    model="gemini-2.0-flash",
    instruction="Fetch relevant data for the user's question.",
    output_key="fetched_data"  # Saves output to state
)

# Step 2: Analyze data
analyzer = Agent(
    name="analyzer",
    model="gemini-2.0-flash",
    instruction="""Analyze the data in {fetched_data}.

    Provide insights and patterns you observe.
    """,
    output_key="analysis"
)

# Step 3: Generate report
reporter = Agent(
    name="reporter",
    model="gemini-2.0-flash",
    instruction="""Based on the analysis in {analysis},
    generate a concise report for the user.

    Include:
    - Key findings
    - Recommendations
    - Next steps
    """
)

# Pipeline executes in order
root_agent = SequentialAgent(
    name="analysis_pipeline",
    sub_agents=[data_fetcher, analyzer, reporter]
)

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

# These run in parallel
fetch_weather = Agent(
    name="weather_fetcher",
    model="gemini-2.0-flash",
    instruction="Get the weather forecast for the user's location.",
    output_key="weather_data",
    tools=[get_weather]
)

fetch_news = Agent(
    name="news_fetcher",
    model="gemini-2.0-flash",
    instruction="Get the top news headlines.",
    output_key="news_data",
    tools=[get_news]
)

fetch_calendar = Agent(
    name="calendar_fetcher",
    model="gemini-2.0-flash",
    instruction="Get today's calendar events.",
    output_key="calendar_data",
    tools=[get_calendar]
)

# Parallel fetching
parallel_fetch = ParallelAgent(
    name="parallel_fetcher",
    sub_agents=[fetch_weather, fetch_news, fetch_calendar]
)

# Synthesize results
synthesizer = Agent(
    name="synthesizer",
    model="gemini-2.0-flash",
    instruction="""Create a morning briefing from:

    Weather: {weather_data}
    News: {news_data}
    Calendar: {calendar_data}

    Make it concise and actionable.
    """
)

# Full pipeline: parallel fetch, then synthesize
root_agent = SequentialAgent(
    name="morning_briefing",
    sub_agents=[parallel_fetch, synthesizer]
)

from google.adk.agents import Agent, LoopAgent

# Agent that checks a condition
checker = Agent(
    name="quality_checker",
    model="gemini-2.0-flash",
    instruction="""Review the draft in {current_draft}.

    If quality is acceptable, respond with APPROVED.
    If improvements needed, respond with specific feedback.
    """,
    output_key="feedback"
)

# Agent that improves based on feedback
improver = Agent(
    name="improver",
    model="gemini-2.0-flash",
    instruction="""Improve the draft based on feedback: {feedback}

    Original draft: {current_draft}

    Create an improved version.
    """,
    output_key="current_draft"
)

# Loop until approved (max 5 iterations)
root_agent = LoopAgent(
    name="refinement_loop",
    max_iterations=5,
    sub_agents=[improver, checker]
)

from google.adk.agents import Agent
from google.adk.tools import AgentTool

# Specialist agent
code_reviewer = Agent(
    name="code_reviewer",
    model="gemini-2.0-flash",
    instruction="""Review the provided code for:
    - Bugs and potential issues
    - Security vulnerabilities
    - Performance concerns
    - Style and readability

    Provide specific, actionable feedback.
    """
)

# Wrap as a tool
code_review_tool = AgentTool(agent=code_reviewer)

# Main agent uses the reviewer as a tool
root_agent = Agent(
    name="coding_assistant",
    model="gemini-2.0-flash",
    instruction="""You help users write code.

    When you generate code, always use the code_reviewer tool
    to check it before presenting to the user.
    """,
    tools=[code_review_tool]
)

from google.adk.agents import Agent
from google.adk.sessions import InMemorySessionService

# Agent that writes to state
greeter = Agent(
    name="greeter",
    model="gemini-2.0-flash",
    instruction="""Greet the user and ask for their name.
    When they provide their name, confirm it.
    """,
    output_key="user_name"  # Saves response to state['user_name']
)

# Agent that reads from state
personalizer = Agent(
    name="personalizer",
    model="gemini-2.0-flash",
    instruction="""Create a personalized welcome message for {user_name}.

    Include:
    - A warm welcome using their name
    - Suggestions for what they can do
    """
)

from google.adk.agents import Agent
from google.adk.sessions import InMemorySessionService
from google.adk.runners import Runner

# Create session service
session_service = InMemorySessionService()

# Create runner
runner = Runner(
    agent=root_agent,
    session_service=session_service,
    app_name="my_app"
)

# Create a session
session = session_service.create_session(
    app_name="my_app",
    user_id="user_123"
)

# Run the agent
async def chat(user_message: str):
    response = await runner.run(
        session_id=session.id,
        user_message=user_message
    )
    return response.text

from google.adk.agents import Agent
from google.adk.agents.callback_context import CallbackContext
from google.genai.types import Content, Part
from typing import Optional

# Logging callback - observes but doesn't modify
async def log_before_model(
    callback_context: CallbackContext,
    llm_request
) -> Optional[any]:
    """Log all LLM requests for debugging and monitoring."""
    print(f"[{callback_context.agent_name}] LLM Request:")
    print(f"  Messages: {len(llm_request.contents)} turns")
    print(f"  Tools: {[t.name for t in (llm_request.tools or [])]}")

    # Return None to allow normal LLM call
    return None


# Caching callback - returns cached response to skip LLM
async def cache_before_model(
    callback_context: CallbackContext,
    llm_request
) -> Optional[any]:
    """Return cached responses for repeated queries."""
    # Generate cache key from request
    cache_key = hash(str(llm_request.contents[-1]))

    cached_response = cache.get(cache_key)
    if cached_response:
        print(f"Cache hit for {cache_key}")
        # Return LlmResponse to skip the actual LLM call
        return cached_response

    # Return None to proceed with LLM call
    return None


# Guardrail callback - blocks dangerous operations
async def guardrail_before_tool(
    callback_context: CallbackContext,
    tool_name: str,
    tool_args: dict
) -> Optional[dict]:
    """Block dangerous tool operations."""

    # Block file deletion
    if tool_name == "delete_file":
        return {
            "status": "error",
            "error_message": "File deletion is not allowed. Please contact an administrator."
        }

    # Block certain paths
    if tool_name in ["read_file", "write_file"]:
        path = tool_args.get("path", "")
        if "/etc/" in path or "/root/" in path:
            return {
                "status": "error",
                "error_message": f"Access to {path} is not permitted."
            }

    # Allow other operations
    return None


# Output filtering callback - modifies responses
async def filter_after_model(
    callback_context: CallbackContext,
    llm_response
) -> Optional[any]:
    """Filter PII from model responses."""
    import re

    # Get the response text
    if llm_response.content and llm_response.content.parts:
        for part in llm_response.content.parts:
            if hasattr(part, 'text') and part.text:
                # Redact email addresses
                part.text = re.sub(
                    r'\b[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Z|a-z]{2,}\b',
                    '[EMAIL REDACTED]',
                    part.text
                )
                # Redact phone numbers
                part.text = re.sub(
                    r'\b\d{3}[-.]?\d{3}[-.]?\d{4}\b',
                    '[PHONE REDACTED]',
                    part.text
                )

    return llm_response


# Register callbacks on the agent
root_agent = Agent(
    name="secure_assistant",
    model="gemini-2.0-flash",
    instruction="You are a helpful assistant.",
    before_model_callback=log_before_model,
    after_model_callback=filter_after_model,
    before_tool_callback=guardrail_before_tool
)

from google.adk.agents import Agent
from google.adk.streaming import LiveRequestQueue, run_live
from google.adk.runners import Runner
import asyncio

# Create a standard agent
agent = Agent(
    name="voice_assistant",
    model="gemini-2.0-flash",  # Must support live API
    instruction="""You are a voice assistant.

    Keep responses concise (1-2 sentences) for natural conversation.
    If interrupted, acknowledge and pivot to the new topic.
    """
)

async def voice_conversation():
    """Run a live voice conversation with the agent."""

    # Create the request queue for sending audio/text
    request_queue = LiveRequestQueue()

    # Start the live session
    async with run_live(
        agent=agent,
        request_queue=request_queue,
        response_modalities=["AUDIO", "TEXT"]  # Get both audio and text
    ) as live_session:

        # Send text message (or audio bytes)
        await request_queue.send_text("Hello, what can you help me with?")

        # Process responses
        async for event in live_session:
            if event.type == "audio":
                # Play audio through speakers
                play_audio(event.audio_data)
            elif event.type == "text":
                # Display text transcript
                print(f"Agent: {event.text}")
            elif event.type == "interrupted":
                # User interrupted - agent will stop and listen
                print("(interrupted)")

# Run the conversation
asyncio.run(voice_conversation())

from fastapi import FastAPI, WebSocket
from google.adk.agents import Agent
from google.adk.streaming import LiveRequestQueue, run_live
import json

app = FastAPI()

agent = Agent(
    name="web_voice_agent",
    model="gemini-2.0-flash",
    instruction="You are a helpful voice assistant for our website."
)

@app.websocket("/ws/voice")
async def voice_websocket(websocket: WebSocket):
    """WebSocket endpoint for voice conversations."""
    await websocket.accept()

    request_queue = LiveRequestQueue()

    async with run_live(
        agent=agent,
        request_queue=request_queue,
        response_modalities=["AUDIO", "TEXT"]
    ) as live_session:

        # Task to receive from browser
        async def receive_from_browser():
            while True:
                data = await websocket.receive_bytes()
                # Forward audio to ADK
                await request_queue.send_audio(data)

        # Task to send to browser
        async def send_to_browser():
            async for event in live_session:
                if event.type == "audio":
                    await websocket.send_bytes(event.audio_data)
                elif event.type == "text":
                    await websocket.send_json({
                        "type": "transcript",
                        "text": event.text
                    })

        # Run both tasks concurrently
        await asyncio.gather(
            receive_from_browser(),
            send_to_browser()
        )

from google.adk.agents import Agent
from google.adk.memory import InMemoryMemoryService, VertexAiMemoryBankService
from google.adk.tools import PreloadMemoryTool, LoadMemoryTool
from google.adk.runners import Runner
from google.adk.sessions import InMemorySessionService

# Development: In-memory (non-persistent)
memory_service = InMemoryMemoryService()

# Production: Vertex AI Memory Bank (persistent, semantic search)
# memory_service = VertexAiMemoryBankService(
#     project="your-project-id",
#     location="us-central1",
#     agent_engine_id="your-agent-engine-id"
# )

# Agent with memory tools
agent = Agent(
    name="memory_agent",
    model="gemini-2.0-flash",
    instruction="""You are a personal assistant with long-term memory.

    At the start of conversations:
    - Use PreloadMemory to recall relevant information about this user
    - Reference past conversations when relevant

    During conversations:
    - Remember important facts the user shares (preferences, decisions, etc.)
    - Use LoadMemory if you need to recall something specific

    Be natural about memory - don't constantly remind users what you remember,
    but do use past context to provide better assistance.
    """,
    tools=[
        PreloadMemoryTool(),  # Auto-loads relevant memories at conversation start
        LoadMemoryTool()       # Agent can explicitly search memories
    ]
)

# Configure runner with memory service
session_service = InMemorySessionService()
runner = Runner(
    agent=agent,
    session_service=session_service,
    memory_service=memory_service,
    app_name="my_app"
)

# After a conversation ends, store it in memory
async def end_session(session_id: str):
    session = await session_service.get_session(session_id)

    # Add session contents to memory
    await memory_service.add_session_to_memory(
        session=session,
        app_name="my_app"
    )

# PreloadMemoryTool automatically runs a search like:
# "What do I know about this user?"

# LoadMemoryTool lets the agent search explicitly:
# Agent decides to call: load_memory(query="user's dietary restrictions")

# First conversation (new user)
user: "I'm allergic to peanuts and prefer vegetarian food."
agent: "Got it! I'll remember that you're allergic to peanuts and prefer vegetarian meals."

# [Session ends, added to memory]

# Second conversation (days later)
user: "Can you recommend a restaurant for tonight?"
agent: "I remember you're vegetarian and allergic to peanuts.
        Let me find restaurants that accommodate those needs..."
# [Agent used PreloadMemoryTool which retrieved past preferences]

from google.adk.agents import Agent
from google.adk.artifacts import InMemoryArtifactService, GcsArtifactService
from google.adk.runners import Runner
from google.adk.sessions import InMemorySessionService
import google.genai.types as types

# Development: In-memory
artifact_service = InMemoryArtifactService()

# Production: Google Cloud Storage
# artifact_service = GcsArtifactService(bucket_name="my-artifacts-bucket")

# Configure runner with artifact service
runner = Runner(
    agent=root_agent,
    session_service=InMemorySessionService(),
    artifact_service=artifact_service,
    app_name="my_app"
)

from google.adk.tools import ToolContext
import google.genai.types as types

async def generate_report(
    context: ToolContext,
    report_type: str,
    data: dict
) -> dict:
    """
    Generate a PDF report and save it as an artifact.

    Args:
        report_type: Type of report ("sales", "inventory", etc.)
        data: Data to include in the report
    """
    # Generate PDF bytes (using your preferred library)
    pdf_bytes = create_pdf_report(report_type, data)

    # Create artifact from bytes
    artifact = types.Part.from_bytes(
        data=pdf_bytes,
        mime_type="application/pdf"
    )

    # Save to artifact service
    # Returns version number (1, 2, 3, etc. for each save)
    version = await context.save_artifact(
        filename=f"{report_type}_report.pdf",
        artifact=artifact
    )

    return {
        "status": "success",
        "message": f"Report saved as {report_type}_report.pdf (version {version})",
        "filename": f"{report_type}_report.pdf"
    }


async def analyze_image(
    context: ToolContext,
    image_description: str
) -> dict:
    """
    Load a previously saved image for analysis.

    Args:
        image_description: Description to identify which image
    """
    # Load the artifact
    image_artifact = await context.load_artifact("uploaded_image.png")

    if not image_artifact:
        return {
            "status": "error",
            "error_message": "No image found. Please upload an image first."
        }

    # Access the binary data
    image_bytes = image_artifact.inline_data.data
    mime_type = image_artifact.inline_data.mime_type

    # Process the image...
    analysis_result = process_image(image_bytes)

    return {
        "status": "success",
        "analysis": analysis_result
    }

# Session-scoped (default): Only accessible within this session
await context.save_artifact("report.pdf", artifact)

# User-scoped: Accessible across all user's sessions
# Prefix filename with "user:"
await context.save_artifact("user:profile_picture.png", artifact)

async def list_user_files(context: ToolContext) -> dict:
    """List all artifacts available in this session."""

    list_response = await context.list_artifacts()

    return {
        "status": "success",
        "files": list_response.filenames
    }

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

# Connect to a filesystem MCP server
filesystem_tools = McpToolset(
    connection_params=StdioConnectionParams(
        server_params=StdioServerParameters(
            command="npx",
            args=[
                "-y",
                "@modelcontextprotocol/server-filesystem",
                "/Users/me/documents"  # Directory to expose
            ]
        )
    ),
    # Only expose specific tools (security best practice)
    tool_filter=["read_file", "list_directory", "search_files"]
)

# Connect to GitHub MCP server
github_tools = McpToolset(
    connection_params=StdioConnectionParams(
        server_params=StdioServerParameters(
            command="npx",
            args=["-y", "@modelcontextprotocol/server-github"],
            env={"GITHUB_TOKEN": os.environ["GITHUB_TOKEN"]}
        )
    ),
    tool_filter=["list_repos", "get_file_contents", "create_issue"]
)

# Agent with MCP tools
agent = Agent(
    name="dev_assistant",
    model="gemini-2.0-flash",
    instruction="""You are a developer assistant with access to:

    1. Local filesystem (read-only, in ~/documents)
       - Use list_directory to explore
       - Use read_file to view file contents
       - Use search_files to find files

    2. GitHub integration
       - Use list_repos to see available repositories
       - Use get_file_contents to read repo files
       - Use create_issue to report bugs or request features

    Always confirm with the user before creating issues.
    """,
    tools=[filesystem_tools, github_tools]
)

from mcp.server.lowlevel import Server
from mcp import types as mcp_types
from google.adk.tools.function_tool import FunctionTool
from google.adk.tools.mcp_tool.conversion_utils import adk_to_mcp_tool_type
import json

# Your ADK tool
def calculate_mortgage(
    principal: float,
    annual_rate: float,
    years: int
) -> dict:
    """Calculate monthly mortgage payment."""
    monthly_rate = annual_rate / 100 / 12
    num_payments = years * 12
    payment = principal * (monthly_rate * (1 + monthly_rate)**num_payments) / \
              ((1 + monthly_rate)**num_payments - 1)
    return {"monthly_payment": round(payment, 2)}

# Wrap as FunctionTool
adk_tool = FunctionTool(calculate_mortgage)

# Create MCP server
app = Server("mortgage-calculator-server")

@app.list_tools()
async def list_tools() -> list[mcp_types.Tool]:
    """Advertise available tools to MCP clients."""
    return [adk_to_mcp_tool_type(adk_tool)]

@app.call_tool()
async def call_tool(name: str, arguments: dict) -> list[mcp_types.Content]:
    """Handle tool invocations from MCP clients."""
    if name == adk_tool.name:
        result = await adk_tool.run_async(args=arguments, tool_context=None)
        return [mcp_types.TextContent(type="text", text=json.dumps(result))]
    raise ValueError(f"Unknown tool: {name}")

# Run server (stdio transport)
if __name__ == "__main__":
    import asyncio
    from mcp.server.stdio import stdio_server
    asyncio.run(app.run(stdio_server()))

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

# Load your OpenAPI spec
openapi_spec = """
openapi: 3.0.0
info:
  title: Pet Store API
  version: 1.0.0
paths:
  /pets:
    get:
      operationId: listPets
      summary: List all pets
      parameters:
        - name: limit
          in: query
          schema:
            type: integer
      responses:
        '200':
          description: A list of pets
  /pets/{petId}:
    get:
      operationId: getPet
      summary: Get a pet by ID
      parameters:
        - name: petId
          in: path
          required: true
          schema:
            type: string
      responses:
        '200':
          description: A pet
"""

# Create toolset from spec
pet_api_tools = OpenAPIToolset(
    spec_str=openapi_spec,
    spec_str_type="yaml"
)

# Agent automatically gets list_pets and get_pet tools
agent = Agent(
    name="pet_store_agent",
    model="gemini-2.0-flash",
    instruction="""You help users find information about pets in our store.

    Use the list_pets tool to show available pets.
    Use the get_pet tool to get details about a specific pet.
    """,
    tools=[pet_api_tools]
)

from google.adk.tools.openapi_tool.openapi_spec_parser.openapi_toolset import OpenAPIToolset
from google.adk.tools.openapi_tool.auth import ApiKeyAuth, BearerAuth, OAuth2Auth

# API Key authentication
toolset = OpenAPIToolset(
    spec_str=openapi_spec,
    spec_str_type="json",
    auth_scheme="apiKey",
    auth_credential=ApiKeyAuth(
        api_key="your-api-key",
        header_name="X-API-Key"
    )
)

# Bearer token authentication
toolset = OpenAPIToolset(
    spec_str=openapi_spec,
    spec_str_type="json",
    auth_scheme="bearer",
    auth_credential=BearerAuth(token="your-bearer-token")
)

# OAuth2 (for complex flows)
toolset = OpenAPIToolset(
    spec_str=openapi_spec,
    spec_str_type="json",
    auth_scheme="oauth2",
    auth_credential=OAuth2Auth(
        client_id="your-client-id",
        client_secret="your-client-secret",
        token_url="https://auth.example.com/token"
    )
)

def search_database(query: str, limit: int = 10) -> dict:
    """Search the product database."""

    # Validate inputs
    if not query or len(query) < 2:
        return {
            "status": "error",
            "error_type": "validation",
            "error_message": "Query must be at least 2 characters."
        }

    if limit < 1 or limit > 100:
        return {
            "status": "error",
            "error_type": "validation",
            "error_message": "Limit must be between 1 and 100."
        }

    try:
        results = database.search(query, limit=limit)
        return {
            "status": "success",
            "results": results,
            "count": len(results)
        }
    except ConnectionError as e:
        return {
            "status": "error",
            "error_type": "connection",
            "error_message": "Database connection failed. Please try again.",
            "retry_after_seconds": 5
        }
    except TimeoutError as e:
        return {
            "status": "error",
            "error_type": "timeout",
            "error_message": "Search timed out. Try a more specific query."
        }
    except Exception as e:
        # Log the actual error for debugging
        logger.exception(f"Unexpected error in search_database: {e}")
        return {
            "status": "error",
            "error_type": "internal",
            "error_message": "An unexpected error occurred. Please try again."
        }

agent = Agent(
    name="resilient_agent",
    model="gemini-2.0-flash",
    instruction="""You are a helpful assistant.

    ## Error Handling

    When a tool returns an error:
    1. Don't apologize excessively - be matter-of-fact
    2. Explain what went wrong in simple terms
    3. Suggest alternatives if available
    4. For "retry_after_seconds" errors, tell the user to wait

    When you're uncertain:
    1. Don't guess or make up information
    2. Tell the user what you don't know
    3. Suggest how they might find the answer

    When user input is unclear:
    1. Ask clarifying questions
    2. Provide examples of valid input
    """,
    tools=[search_database]
)

async def handle_tool_errors(
    callback_context: CallbackContext,
    tool_name: str,
    tool_result: dict
) -> Optional[dict]:
    """Centralized error handling for all tools."""

    if tool_result.get("status") == "error":
        error_type = tool_result.get("error_type", "unknown")

        # Log all errors
        logger.error(
            f"Tool error: {tool_name}",
            extra={
                "error_type": error_type,
                "error_message": tool_result.get("error_message"),
                "session_id": callback_context.session.id
            }
        )

        # Track error metrics
        metrics.increment(
            "tool_errors",
            tags={"tool": tool_name, "error_type": error_type}
        )

        # For certain errors, modify the response
        if error_type == "authentication":
            return {
                "status": "error",
                "error_message": "Session expired. Please log in again."
            }

    return tool_result  # Return original result

agent = Agent(
    name="monitored_agent",
    model="gemini-2.0-flash",
    instruction="...",
    after_tool_callback=handle_tool_errors
)

from google.adk.agents import Agent, SequentialAgent

# Primary data source
primary_search = Agent(
    name="primary_search",
    model="gemini-2.0-flash",
    instruction="Search the primary database.",
    tools=[primary_db_search],
    output_key="search_results"
)

# Fallback if primary fails
fallback_search = Agent(
    name="fallback_search",
    model="gemini-2.0-flash",
    instruction="""Check if {search_results} contains an error.
    If so, search the backup database instead.
    If primary succeeded, just pass through the results.
    """,
    tools=[backup_db_search],
    output_key="final_results"
)

# Pipeline with fallback
search_pipeline = SequentialAgent(
    name="resilient_search",
    sub_agents=[primary_search, fallback_search]
)

// tests/weather_agent.test.json
{
  "name": "Weather Agent Tests",
  "description": "Test cases for the weather agent",
  "eval_cases": [
    {
      "name": "basic_weather_query",
      "conversation": [
        {
          "role": "user",
          "content": "What's the weather in New York?"
        }
      ],
      "expected_tool_calls": [
        {
          "tool_name": "get_weather",
          "arguments": {"city": "New York"}
        }
      ],
      "expected_response_contains": ["New York", "temperature"],
      "reference_response": "The weather in New York is currently sunny with a temperature of 22°C."
    },
    {
      "name": "unknown_city",
      "conversation": [
        {
          "role": "user",
          "content": "What's the weather in Atlantis?"
        }
      ],
      "expected_tool_calls": [
        {
          "tool_name": "get_weather",
          "arguments": {"city": "Atlantis"}
        }
      ],
      "expected_response_contains": ["not available", "don't have"]
    }
  ]
}

# Run all tests for an agent
adk eval my_agent tests/

# Run specific test file
adk eval my_agent tests/weather_agent.test.json

# Show detailed results
adk eval my_agent tests/ --print_detailed_results

import pytest
from google.adk.evaluation.agent_evaluator import AgentEvaluator

@pytest.mark.asyncio
async def test_weather_agent():
    results = await AgentEvaluator.evaluate(
        agent_module="my_agent",
        eval_dataset_file_path_or_dir="tests/weather_agent.test.json"
    )

    # Check overall pass rate
    assert results.pass_rate >= 0.9, f"Pass rate too low: {results.pass_rate}"

    # Check specific metrics
    for case_result in results.case_results:
        assert case_result.tool_trajectory_score >= 0.8
        assert case_result.response_match_score >= 0.7

adk web  # Navigate to Evaluation tab

{
  "name": "multi_turn_conversation",
  "conversation": [
    {"role": "user", "content": "I want to book a flight to Paris."},
    {"role": "assistant", "content": "I'd be happy to help you book a flight to Paris. What date are you looking to travel?"},
    {"role": "user", "content": "Next Friday."},
    {"role": "assistant", "content": "And which city will you be departing from?"},
    {"role": "user", "content": "New York."}
  ],
  "expected_tool_calls": [
    {
      "tool_name": "search_flights",
      "arguments": {
        "origin": "New York",
        "destination": "Paris"
      }
    }
  ]
}

# .github/workflows/agent-tests.yml
name: Agent Tests

on: [push, pull_request]

jobs:
  test:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v3

      - name: Set up Python
        uses: actions/setup-python@v4
        with:
          python-version: '3.11'

      - name: Install dependencies
        run: |
          pip install google-adk pytest pytest-asyncio
          pip install -r requirements.txt

      - name: Run agent evaluations
        env:
          GOOGLE_API_KEY: ${{ secrets.GOOGLE_API_KEY }}
        run: |
          adk eval my_agent tests/ --print_detailed_results

      - name: Run pytest
        run: pytest tests/ -v

agents/python/customer-service/
├── agent.py           # Main coordinator
├── agents/
│   ├── billing.py     # Billing specialist
│   ├── technical.py   # Technical support
│   └── general.py     # General inquiries
├── tools/
│   ├── crm.py         # Customer data tools
│   └── ticketing.py   # Support ticket tools
└── prompts/
    └── system.py      # Shared prompt templates

agents/python/data-science/
├── agent.py
├── tools/
│   ├── pandas_tools.py    # DataFrame operations
│   ├── viz_tools.py       # Chart generation
│   └── stats_tools.py     # Statistical analysis
└── notebooks/             # Example analyses

agents/python/travel-concierge/
├── agent.py               # Coordinator
├── agents/
│   ├── flights.py         # Flight search/booking
│   ├── hotels.py          # Hotel recommendations
│   ├── activities.py      # Local activities
│   └── itinerary.py       # Trip planning
└── tools/
    ├── search_apis.py     # External API integrations
    └── calendar.py        # Schedule management

# Clone the samples repository
git clone https://github.com/google/adk-samples.git
cd adk-samples

# Navigate to a sample
cd agents/python/customer-service

# Install dependencies
pip install -r requirements.txt

# Set up environment
cp .env.example .env
# Edit .env with your API key

# Run with web UI
adk web

# Or run in terminal
adk run .

from google.adk.agents import Agent
from google.genai.types import GenerateContentConfig

agent = Agent(
    name="creative_writer",
    model="gemini-2.0-flash",
    instruction="Write creative stories based on user prompts.",
    generate_content_config=GenerateContentConfig(
        temperature=0.9,        # Higher = more creative
        max_output_tokens=2000,
        top_p=0.95,
        top_k=40
    )
)

from pydantic import BaseModel
from typing import List

class TaskList(BaseModel):
    tasks: List[str]
    priority: str
    estimated_hours: float

agent = Agent(
    name="task_planner",
    model="gemini-2.0-flash",
    instruction="Create task lists based on user goals.",
    output_schema=TaskList  # Enforces JSON output matching schema
)

from google.adk.agents import Agent
from google.adk.planners import BuiltInPlanner

agent = Agent(
    name="complex_reasoner",
    model="gemini-2.0-flash-thinking",  # Thinking-enabled model
    instruction="Solve complex problems step by step.",
    planner=BuiltInPlanner(
        thinking_budget=1024,      # Tokens for thinking
        include_thoughts=True      # Include reasoning in response
    )
)

# Install the Vertex AI SDK
pip install google-cloud-aiplatform

# Deploy
adk deploy --project=your-project-id --region=us-central1