# Google Agent Development Kit (ADK) - Python

The Google Agent Development Kit (ADK) is a flexible, code-first Python framework for building, evaluating, and deploying sophisticated AI agents. It provides a modular architecture that enables developers to create everything from simple single-agent applications to complex multi-agent systems. While optimized for Google's Gemini models, ADK is model-agnostic and compatible with various LLM providers.

ADK simplifies agent development through a rich tool ecosystem including built-in tools, custom Python functions, OpenAPI integrations, and MCP (Model Context Protocol) support. The framework offers multiple agent orchestration patterns (sequential, parallel, loop), comprehensive session and memory management, artifact handling, and plugin-based extensibility. It supports deployment anywhere from local development to Cloud Run and Vertex AI Agent Engine.

## Core Agent API

### Agent (LlmAgent)

The `Agent` class (alias for `LlmAgent`) is the primary building block for creating AI agents. It wraps an LLM model with instructions, tools, and optional sub-agents for hierarchical orchestration.

```python
from google.adk.agents import Agent
from google.adk.tools import google_search

# Create a simple agent with tools
root_agent = Agent(
    name="search_assistant",
    model="gemini-2.5-flash",
    instruction="You are a helpful assistant. Answer user questions using Google Search when needed.",
    description="An assistant that can search the web.",
    tools=[google_search]
)

# Create an agent with custom functions as tools
def get_weather(city: str) -> str:
    """Get the current weather for a city.

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

    Returns:
        Weather information string.
    """
    return f"The weather in {city} is sunny, 72°F"

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

### Multi-Agent Systems

ADK supports hierarchical multi-agent systems where a coordinator agent can delegate to specialized sub-agents. The framework automatically handles agent transfer and conversation context.

```python
from google.adk.agents import LlmAgent

# Define specialized agents
greeter = LlmAgent(
    name="greeter",
    model="gemini-2.5-flash",
    instruction="You warmly greet users and introduce the service.",
    description="Handles greetings and introductions"
)

researcher = LlmAgent(
    name="researcher",
    model="gemini-2.5-flash",
    instruction="You research topics and provide detailed information.",
    description="Handles research queries"
)

# Create coordinator with sub-agents
coordinator = LlmAgent(
    name="coordinator",
    model="gemini-2.5-flash",
    instruction="""You are the main coordinator. Route users to:
    - greeter: for hellos and introductions
    - researcher: for information queries""",
    description="Routes requests to appropriate specialists",
    sub_agents=[greeter, researcher]
)
```

## Runner API

### InMemoryRunner

The `InMemoryRunner` provides a lightweight runner for testing and development. It uses in-memory storage for sessions, artifacts, and memory.

```python
import asyncio
from google.adk.agents import Agent
from google.adk.runners import InMemoryRunner
from google.genai import types

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

# Initialize runner
runner = InMemoryRunner(agent=agent, app_name="my_app")

async def main():
    # Quick debug method for testing
    events = await runner.run_debug("What is 2 + 2?")

    # Or use full async API for production
    async for event in runner.run_async(
        user_id="user_123",
        session_id="session_456",
        new_message=types.UserContent(parts=[types.Part(text="Hello!")])
    ):
        if event.content and event.content.parts:
            for part in event.content.parts:
                if part.text:
                    print(f"Agent: {part.text}")

asyncio.run(main())
```

### Runner with Custom Services

For production environments, use the full `Runner` class with custom session, artifact, and memory services.

```python
import asyncio
from google.adk.agents import Agent
from google.adk.runners import Runner
from google.adk.apps.app import App
from google.adk.sessions.in_memory_session_service import InMemorySessionService
from google.adk.artifacts.in_memory_artifact_service import InMemoryArtifactService
from google.adk.memory.in_memory_memory_service import InMemoryMemoryService
from google.genai import types

agent = Agent(
    name="my_agent",
    model="gemini-2.0-flash",
    instruction="You are helpful."
)

# Create an App for better configuration
app = App(
    name="my_application",
    root_agent=agent
)

# Initialize runner with services
runner = Runner(
    app=app,
    session_service=InMemorySessionService(),
    artifact_service=InMemoryArtifactService(),
    memory_service=InMemoryMemoryService()
)

async def chat():
    # Create a session first
    session = await runner.session_service.create_session(
        app_name="my_application",
        user_id="user_1",
        session_id="session_1"
    )

    # Run agent
    async for event in runner.run_async(
        user_id="user_1",
        session_id="session_1",
        new_message=types.UserContent(parts=[types.Part(text="Hi there!")])
    ):
        print(f"Event from {event.author}: {event.content}")

asyncio.run(chat())
```

## Session Management

### InMemorySessionService

The session service manages conversation state across multiple interactions. Sessions store events, state, and can be persisted.

```python
import asyncio
from google.adk.sessions.in_memory_session_service import InMemorySessionService
from google.adk.sessions.base_session_service import GetSessionConfig

session_service = InMemorySessionService()

async def manage_sessions():
    # Create a new session with initial state
    session = await session_service.create_session(
        app_name="my_app",
        user_id="user_123",
        session_id="conversation_1",
        state={"user_preference": "dark_mode"}
    )
    print(f"Created session: {session.id}")

    # Retrieve session with optional filtering
    session = await session_service.get_session(
        app_name="my_app",
        user_id="user_123",
        session_id="conversation_1",
        config=GetSessionConfig(num_recent_events=10)  # Get last 10 events
    )

    # List all sessions for a user
    response = await session_service.list_sessions(
        app_name="my_app",
        user_id="user_123"
    )
    for s in response.sessions:
        print(f"Session: {s.id}, Last update: {s.last_update_time}")

    # Delete session
    await session_service.delete_session(
        app_name="my_app",
        user_id="user_123",
        session_id="conversation_1"
    )

asyncio.run(manage_sessions())
```

## Tools

### FunctionTool

Any Python function can be used as a tool. ADK automatically extracts the function signature and docstring to create the tool declaration for the LLM.

```python
from google.adk.agents import Agent
from google.adk.tools.tool_context import ToolContext

# Simple function tool
def calculate(expression: str) -> str:
    """Evaluate a mathematical expression.

    Args:
        expression: A math expression like "2 + 3 * 4"

    Returns:
        The calculated result as a string.
    """
    try:
        result = eval(expression)  # Note: Use safe evaluation in production
        return str(result)
    except Exception as e:
        return f"Error: {e}"

# Function with tool context for state access
def remember_fact(fact: str, tool_context: ToolContext) -> str:
    """Store a fact in the session state.

    Args:
        fact: The fact to remember.

    Returns:
        Confirmation message.
    """
    facts = tool_context.state.get("facts", [])
    facts.append(fact)
    tool_context.state["facts"] = facts
    return f"Remembered: {fact}. Total facts stored: {len(facts)}"

# Async function tool
async def fetch_data(url: str) -> dict:
    """Fetch data from a URL.

    Args:
        url: The URL to fetch data from.

    Returns:
        The fetched data as a dictionary.
    """
    import aiohttp
    async with aiohttp.ClientSession() as session:
        async with session.get(url) as response:
            return await response.json()

agent = Agent(
    name="tool_demo",
    model="gemini-2.0-flash",
    instruction="Use tools to help users.",
    tools=[calculate, remember_fact, fetch_data]
)
```

### GoogleSearchTool

Built-in Google Search tool that integrates directly with Gemini models for real-time web search capabilities.

```python
from google.adk.agents import Agent
from google.adk.tools import google_search

# Agent with Google Search capability
search_agent = Agent(
    name="search_agent",
    model="gemini-2.0-flash",
    instruction="""You help users find information online.
    Use Google Search to find current, accurate information.""",
    tools=[google_search]
)
```

### McpToolset

Integration with Model Context Protocol (MCP) servers for accessing external tools and services.

```python
import asyncio
from google.adk.agents import LlmAgent
from google.adk.tools.mcp_tool.mcp_toolset import McpToolset
from google.adk.tools.mcp_tool.mcp_session_manager import StdioConnectionParams

# Connect to an MCP filesystem server
filesystem_toolset = McpToolset(
    connection_params=StdioConnectionParams(
        command="npx",
        args=["-y", "@modelcontextprotocol/server-filesystem", "/tmp"],
        timeout=30.0
    ),
    tool_filter=["read_file", "list_directory"]  # Optional: filter specific tools
)

# Agent with MCP tools
file_agent = LlmAgent(
    name="file_agent",
    model="gemini-2.0-flash",
    instruction="Help users explore and read files.",
    tools=[filesystem_toolset]
)

# Remember to close the toolset when done
async def cleanup():
    await filesystem_toolset.close()
```

### OpenAPIToolset

Generate tools automatically from OpenAPI specifications for REST API integration.

```python
from google.adk.agents import Agent
from google.adk.tools.openapi_tool.openapi_spec_parser.openapi_toolset import OpenAPIToolset
from google.adk.auth.auth_schemes import AuthSchemeApiKey, ApiKeyLocation
from google.adk.auth.auth_credential import AuthCredential, AuthCredentialTypes

# OpenAPI spec as string
openapi_spec = """
{
  "openapi": "3.0.0",
  "info": {"title": "Weather API", "version": "1.0"},
  "servers": [{"url": "https://api.weather.example.com"}],
  "paths": {
    "/weather/{city}": {
      "get": {
        "operationId": "getWeather",
        "summary": "Get weather for a city",
        "parameters": [
          {"name": "city", "in": "path", "required": true, "schema": {"type": "string"}}
        ],
        "responses": {"200": {"description": "Weather data"}}
      }
    }
  }
}
"""

# Create toolset with authentication
weather_toolset = OpenAPIToolset(
    spec_str=openapi_spec,
    spec_str_type="json",
    auth_scheme=AuthSchemeApiKey(
        in_location=ApiKeyLocation.HEADER,
        name="X-API-Key"
    ),
    auth_credential=AuthCredential(
        auth_type=AuthCredentialTypes.API_KEY,
        api_key="your-api-key-here"
    )
)

# Use in agent
api_agent = Agent(
    name="api_agent",
    model="gemini-2.0-flash",
    instruction="Help users check weather using the API.",
    tools=[weather_toolset]
)
```

## Orchestration Agents

### SequentialAgent

Runs sub-agents in sequence, passing context between them. Useful for pipeline-style workflows.

```python
from google.adk.agents import LlmAgent
from google.adk.agents.sequential_agent import SequentialAgent

# Define pipeline stages
data_collector = LlmAgent(
    name="collector",
    model="gemini-2.0-flash",
    instruction="Collect and organize the user's requirements."
)

analyzer = LlmAgent(
    name="analyzer",
    model="gemini-2.0-flash",
    instruction="Analyze the collected requirements and identify key points."
)

reporter = LlmAgent(
    name="reporter",
    model="gemini-2.0-flash",
    instruction="Generate a final report based on the analysis."
)

# Create sequential pipeline
pipeline = SequentialAgent(
    name="analysis_pipeline",
    sub_agents=[data_collector, analyzer, reporter]
)
```

### ParallelAgent

Runs sub-agents concurrently in isolated branches. Useful for getting multiple perspectives or parallel processing.

```python
from google.adk.agents import LlmAgent
from google.adk.agents.parallel_agent import ParallelAgent

# Define parallel workers
optimist = LlmAgent(
    name="optimist",
    model="gemini-2.0-flash",
    instruction="Analyze the topic with an optimistic perspective."
)

pessimist = LlmAgent(
    name="pessimist",
    model="gemini-2.0-flash",
    instruction="Analyze the topic identifying potential risks and concerns."
)

realist = LlmAgent(
    name="realist",
    model="gemini-2.0-flash",
    instruction="Provide a balanced, factual analysis."
)

# Run all perspectives in parallel
multi_perspective = ParallelAgent(
    name="perspective_analyzer",
    sub_agents=[optimist, pessimist, realist]
)
```

### LoopAgent

Runs sub-agents in a loop until an exit condition is met or max iterations reached.

```python
from google.adk.agents import LlmAgent
from google.adk.agents.loop_agent import LoopAgent

# Define iterative refinement agents
drafter = LlmAgent(
    name="drafter",
    model="gemini-2.0-flash",
    instruction="Draft or improve the current document based on feedback."
)

reviewer = LlmAgent(
    name="reviewer",
    model="gemini-2.0-flash",
    instruction="""Review the draft and provide feedback.
    If the document is satisfactory, call escalate to finish."""
)

# Create refinement loop (max 5 iterations)
refinement_loop = LoopAgent(
    name="document_refiner",
    max_iterations=5,
    sub_agents=[drafter, reviewer]
)
```

## Plugins

### BasePlugin

Plugins provide hooks into the agent execution lifecycle for logging, monitoring, caching, or modifying behavior.

```python
from google.adk.plugins.base_plugin import BasePlugin
from google.adk.tools.base_tool import BaseTool
from google.adk.agents.base_agent import BaseAgent
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.events.event import Event
from typing import Any, Optional
import time

class MonitoringPlugin(BasePlugin):
    """Plugin that monitors and logs all agent activity."""

    def __init__(self):
        super().__init__(name="monitoring_plugin")
        self.tool_calls = []
        self.model_calls = []

    async def before_agent_callback(
        self, *, agent: BaseAgent, callback_context: CallbackContext
    ) -> Optional[Any]:
        print(f"[Monitor] Agent '{agent.name}' starting...")
        return None  # Continue normally

    async def after_agent_callback(
        self, *, agent: BaseAgent, callback_context: CallbackContext
    ) -> Optional[Any]:
        print(f"[Monitor] Agent '{agent.name}' completed")
        return None

    async def before_tool_callback(
        self, *, tool: BaseTool, tool_args: dict[str, Any], tool_context
    ) -> Optional[dict]:
        start = time.time()
        print(f"[Monitor] Calling tool '{tool.name}' with args: {tool_args}")
        self.tool_calls.append({"tool": tool.name, "start": start})
        return None  # Proceed with tool call

    async def after_tool_callback(
        self, *, tool: BaseTool, tool_args: dict[str, Any],
        tool_context, result: dict
    ) -> Optional[dict]:
        duration = time.time() - self.tool_calls[-1]["start"]
        print(f"[Monitor] Tool '{tool.name}' completed in {duration:.2f}s")
        return None  # Use original result

    async def before_model_callback(
        self, *, callback_context: CallbackContext, llm_request: LlmRequest
    ) -> Optional[LlmResponse]:
        print(f"[Monitor] Sending request to model: {llm_request.model}")
        return None  # Proceed with model call

    async def on_event_callback(
        self, *, invocation_context, event: Event
    ) -> Optional[Event]:
        print(f"[Monitor] Event from '{event.author}': {event.id}")
        return None  # Use original event

# Use plugin in runner
from google.adk.runners import Runner
from google.adk.apps.app import App

app = App(
    name="monitored_app",
    root_agent=my_agent,
    plugins=[MonitoringPlugin()]
)
```

## Events and State

### Event

Events represent actions in an agent conversation, including user messages, agent responses, and tool calls.

```python
from google.adk.events.event import Event
from google.adk.events.event_actions import EventActions
from google.genai import types

# Events are typically created by the framework, but can be inspected:
def process_events(events: list[Event]):
    for event in events:
        # Check event author
        if event.author == "user":
            print(f"User said: {event.content}")
        else:
            print(f"Agent '{event.author}' responded")

        # Check for function calls
        func_calls = event.get_function_calls()
        for call in func_calls:
            print(f"Tool called: {call.name} with {call.args}")

        # Check for function responses
        func_responses = event.get_function_responses()
        for response in func_responses:
            print(f"Tool response: {response.name} -> {response.response}")

        # Check if final response
        if event.is_final_response():
            print("This is the final response")

        # Access state changes
        if event.actions.state_delta:
            print(f"State changed: {event.actions.state_delta}")
```

### State Management

State can be managed at session, user, and app levels through the ToolContext or CallbackContext.

```python
from google.adk.tools.tool_context import ToolContext

def stateful_tool(query: str, tool_context: ToolContext) -> str:
    """A tool that uses and modifies session state.

    Args:
        query: The user's query.

    Returns:
        Response based on accumulated state.
    """
    # Read state (session-level by default)
    count = tool_context.state.get("query_count", 0)
    history = tool_context.state.get("query_history", [])

    # Update state
    count += 1
    history.append(query)
    tool_context.state["query_count"] = count
    tool_context.state["query_history"] = history

    # User-level state (persists across sessions)
    tool_context.state["user:total_queries"] = (
        tool_context.state.get("user:total_queries", 0) + 1
    )

    # App-level state (shared across all users)
    tool_context.state["app:global_query_count"] = (
        tool_context.state.get("app:global_query_count", 0) + 1
    )

    return f"Query #{count} recorded. Total queries: {len(history)}"
```

## Memory Service

### InMemoryMemoryService

Memory service enables agents to recall information from past sessions using keyword-based search.

```python
import asyncio
from google.adk.memory.in_memory_memory_service import InMemoryMemoryService
from google.adk.sessions.session import Session

memory_service = InMemoryMemoryService()

async def use_memory():
    # Add a session to memory (typically done via callback_context)
    session = Session(
        app_name="my_app",
        user_id="user_123",
        id="session_1"
    )
    # ... session would have events added during conversation
    await memory_service.add_session_to_memory(session)

    # Search memory
    results = await memory_service.search_memory(
        app_name="my_app",
        user_id="user_123",
        query="previous conversation about weather"
    )

    for memory in results.memories:
        print(f"Found: {memory.content} (from {memory.author})")

asyncio.run(use_memory())
```

## Code Execution

### BuiltInCodeExecutor

Enable Gemini 2.0+ models to execute Python code directly for calculations and data processing.

```python
from google.adk.agents import Agent
from google.adk.code_executors.built_in_code_executor import BuiltInCodeExecutor

# Agent with code execution capability
code_agent = Agent(
    name="code_executor",
    model="gemini-2.0-flash",
    instruction="""You are a data analyst. When users ask for calculations
    or data analysis, write and execute Python code to solve their problems.""",
    code_executor=BuiltInCodeExecutor()
)
```

## Summary

The Google Agent Development Kit provides a comprehensive framework for building AI agents with LLMs. The key integration patterns include: (1) **Single agents** with custom tools for straightforward tasks, (2) **Multi-agent hierarchies** with coordinator agents delegating to specialists, (3) **Orchestration patterns** using Sequential, Parallel, and Loop agents for complex workflows, and (4) **External integrations** via OpenAPI toolsets and MCP servers.

For production deployments, ADK supports custom session services (Cloud SQL, Spanner), artifact storage (Cloud Storage), and memory services for long-term context. The plugin system enables cross-cutting concerns like logging, monitoring, and caching. Applications can be deployed as containers on Cloud Run or scaled with Vertex AI Agent Engine. The framework's event-driven architecture makes it suitable for both synchronous request-response patterns and streaming real-time interactions through the `run_live` API for audio/video applications.