--- 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 | Tool | Description | |---|---| | `google_search` | Google Search grounding | | `code_execution` | Sandboxed Python execution | | `vertex_ai_search` | Enterprise search via Vertex AI | | `function_tool` | Wrap any Python function as a tool | | `MCPToolset` | Wrap any MCP server as ADK tools | | `long_running_tool` | Async tools with deferred results | | `transfer_to_agent` | Delegate to sub-agents | ## Patterns ### 1. Basic Agent with Function Tools ```python 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 ```python 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 ```python 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 ```python 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 ```python # 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](https://google.github.io/adk-docs/) - [Google ADK GitHub](https://github.com/google/adk-python) - [Gemini API](https://ai.google.dev/gemini-api/docs) - [A2A Protocol](https://google.github.io/A2A/)