name: google-adk description: Google Agent Development Kit (ADK) patterns for multi-agent orchestration with Gemini. Use when building agent hierarchies, using MCP tools, implementing A2A protocol, or deploying agents on Google Cloud. tags: [google, adk, agents, multi-agent]

Google Agent Development Kit (ADK)

Build multi-agent AI applications using Google's open-source ADK framework with Gemini models, MCP tool support, and Agent-to-Agent (A2A) protocol.

When to Use

• Building hierarchical multi-agent systems with parent-child delegation
• Connecting agents to MCP servers for enterprise tool access
• Implementing Agent-to-Agent (A2A) protocol for cross-framework interop
• Deploying agents to Cloud Run or Cloud Functions
• Using Gemini models with function calling, code execution, and search grounding

Built-in Tools

Patterns

1. Basic Agent with Function Tools

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

def get_weather(city: str) -> dict:
    """Get current weather for a city.

    Args:
        city: City name (e.g., 'Ho Chi Minh City')
    """
    return {"city": city, "temperature": 32, "condition": "sunny"}

agent = Agent(
    model="gemini-2.0-flash",
    name="weather_agent",
    instruction="Help users check weather conditions.",
    tools=[get_weather],
)

session_service = InMemorySessionService()
runner = Runner(agent=agent, app_name="weather_app", session_service=session_service)

async def main():
    session = await session_service.create_session(app_name="weather_app", user_id="user1")
    response = await runner.run(user_id="user1", session_id=session.id, new_message="Weather in Hanoi?")
    print(response)

2. Multi-Agent Hierarchy

from google.adk.agents import Agent

# Specialist agents
finance_agent = Agent(
    model="gemini-2.0-flash",
    name="finance_specialist",
    instruction="Handle finance and accounting queries. Query financial data and generate reports.",
    tools=[query_gl_accounts, get_financial_report],
)

hr_agent = Agent(
    model="gemini-2.0-flash",
    name="hr_specialist",
    instruction="Handle HR queries. Look up employee info and process requests.",
    tools=[get_employee, submit_leave_request],
)

# Supervisor delegates to specialists
supervisor = Agent(
    model="gemini-2.0-flash",
    name="enterprise_assistant",
    instruction="""Route user requests to the appropriate specialist:
    - Finance/accounting questions -> finance_specialist
    - HR/employee questions -> hr_specialist
    For general questions, answer directly.""",
    sub_agents=[finance_agent, hr_agent],
)

3. MCP Server Integration

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

# Connect to Salesforce MCP server
salesforce_mcp = MCPToolset(
    connection_params=StdioServerParameters(
        command="npx",
        args=["-y", "@salesforce/mcp-server"],
        env={"SF_ACCESS_TOKEN": "...", "SF_INSTANCE_URL": "https://myorg.salesforce.com"},
    )
)

# Connect to database MCP server
db_mcp = MCPToolset(
    connection_params=StdioServerParameters(
        command="npx",
        args=["-y", "@modelcontextprotocol/server-postgres", "postgresql://..."],
    )
)

agent = Agent(
    model="gemini-2.0-flash",
    name="data_agent",
    instruction="Answer questions using CRM and database data.",
    tools=[salesforce_mcp, db_mcp],
)

4. Callbacks and Lifecycle Hooks

from google.adk.agents import Agent

async def before_tool_callback(tool_name: str, tool_input: dict, context):
    """Audit log all tool calls."""
    print(f"[AUDIT] Tool: {tool_name}, Input: {tool_input}")
    return None  # Return None to continue, or a dict to override

async def after_tool_callback(tool_name: str, tool_output, context):
    """Post-process tool results."""
    print(f"[AUDIT] Tool: {tool_name}, Output length: {len(str(tool_output))}")
    return None

agent = Agent(
    model="gemini-2.0-flash",
    name="audited_agent",
    instruction="Help users with data queries.",
    tools=[query_data],
    before_tool_callback=before_tool_callback,
    after_tool_callback=after_tool_callback,
)

5. Deploy to Cloud Run

# main.py - Cloud Run deployment
from google.adk.agents import Agent
from google.adk.runners import Runner
from google.adk.sessions import DatabaseSessionService
from fastapi import FastAPI

app = FastAPI()

agent = Agent(model="gemini-2.0-flash", name="my_agent", instruction="...")
session_service = DatabaseSessionService(db_url="postgresql://...")
runner = Runner(agent=agent, app_name="my_app", session_service=session_service)

@app.post("/chat")
async def chat(user_id: str, message: str, session_id: str = None):
    if not session_id:
        session = await session_service.create_session(app_name="my_app", user_id=user_id)
        session_id = session.id
    response = await runner.run(user_id=user_id, session_id=session_id, new_message=message)
    return {"response": str(response), "session_id": session_id}

Anti-Patterns

• Using synchronous tools for I/O-bound operations -- use async functions
• Deep nesting of sub-agents (> 3 levels) -- flatten hierarchy where possible
• Not using session persistence for production -- InMemorySessionService loses state on restart
• Hardcoding model names -- use configuration for easy model switching

References

• Google ADK Documentation
• Google ADK GitHub
• Gemini API
• A2A Protocol