my_agent/
├── my_agent/
│   ├── __init__.py          # Must import agent module
│   ├── agent.py             # Defines root_agent (entry point)
│   ├── prompts.py           # Instruction strings (optional)
│   ├── tools.py             # Custom tool functions (optional)
│   ├── sub_agents/          # Sub-agent packages (optional)
│   └── shared_libraries/    # Callbacks, utilities (optional)
├── tests/
│   └── test_agent.py
├── pyproject.toml
└── .env                     # GOOGLE_API_KEY or GOOGLE_CLOUD_PROJECT

# my_agent/__init__.py
from . import agent

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

root_agent = Agent(
    name="my_agent",
    model="gemini-2.5-flash",
    description="Brief description for agent discovery",
    instruction="Detailed system prompt...",
    tools=[...],
)

[project]
name = "my-agent"
version = "0.1.0"
requires-python = ">=3.10"
dependencies = ["google-adk"]

[build-system]
requires = ["setuptools"]
build-backend = "setuptools.build_meta"

from google.adk.agents import LlmAgent

agent = LlmAgent(
    name="assistant",
    model="gemini-2.5-flash",
    description="General assistant",
    instruction="""You are a helpful assistant.
    Use the search tool when you need current information.""",
    tools=[search_tool],
    output_schema=ResponseModel,   # Optional structured output
    output_key="response",         # State key for output
    generate_content_config=types.GenerateContentConfig(
        temperature=0.7,
    ),
)

from google.adk.agents import SequentialAgent

pipeline = SequentialAgent(
    name="research_pipeline",
    description="Research then summarize",
    sub_agents=[
        researcher_agent,    # Step 1: writes to state["research"]
        summarizer_agent,    # Step 2: reads state["research"]
    ],
)

from google.adk.agents import ParallelAgent

parallel = ParallelAgent(
    name="multi_channel",
    description="Send to all channels simultaneously",
    sub_agents=[
        email_agent,
        slack_agent,
        calendar_agent,
    ],
)

from google.adk.agents import LoopAgent

loop = LoopAgent(
    name="refinement_loop",
    description="Iteratively refine output",
    sub_agents=[writer_agent, critic_agent],
    max_iterations=3,
)

# The checker agent uses tool_context.actions.escalate = True to stop
def check_quality(score: float, tool_context: ToolContext) -> str:
    """Check if quality meets threshold."""
    if score >= 0.9:
        tool_context.actions.escalate = True
        return "Quality threshold met, stopping loop."
    return "Quality below threshold, continuing refinement."

checker = Agent(
    name="checker",
    model="gemini-2.5-flash",
    instruction="Evaluate the output quality and call check_quality.",
    tools=[check_quality],
)

loop = LoopAgent(
    name="quality_loop",
    sub_agents=[generator_agent, checker],
)

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

    Args:
        city: The city name to look up weather for.
        units: Temperature units - 'celsius' or 'fahrenheit'.

    Returns:
        dict with temperature, conditions, and humidity.
    """
    # Implementation
    return {"temperature": 22, "conditions": "sunny", "humidity": 45}

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

from google.adk.tools import ToolContext

def add_to_cart(item: str, quantity: int, tool_context: ToolContext) -> dict:
    """Add an item to the shopping cart."""
    cart = tool_context.state.get("cart", [])
    cart.append({"item": item, "quantity": quantity})
    tool_context.state["cart"] = cart
    return {"status": "added", "cart_size": len(cart)}

from google.adk.tools.agent_tool import AgentTool

specialist = Agent(
    name="code_reviewer",
    model="gemini-2.5-pro",
    instruction="Review code for bugs and best practices.",
)

coordinator = Agent(
    name="coordinator",
    model="gemini-2.5-flash",
    instruction="Coordinate tasks. Use code_reviewer for code reviews.",
    tools=[AgentTool(agent=specialist)],
)

from google.adk.tools import google_search

agent = Agent(
    name="researcher",
    tools=[google_search],
)

from google.adk.tools.mcp_tool import MCPToolset, StdioConnectionParams
from mcp import StdioServerParameters

agent = Agent(
    name="db_agent",
    tools=[
        MCPToolset(
            connection_params=StdioConnectionParams(
                server_params=StdioServerParameters(
                    command="npx",
                    args=["-y", "some-mcp-server"],
                ),
            ),
        ),
    ],
)

def before_tool(tool, args, tool_context) -> dict | None:
    """Return dict to skip tool execution with that response."""
    if tool.name == "approve_discount" and args.get("value", 0) > 50:
        return {"status": "rejected", "reason": "Discount too large"}
    return None  # Proceed normally

agent = Agent(
    name="guarded_agent",
    model="gemini-2.5-flash",
    instruction="...",
    before_tool_callback=before_tool,
)

researcher = Agent(name="researcher", output_key="findings", output_schema=ResearchOutput, ...)
writer = Agent(name="writer", instruction="Write report based on state['findings'].", ...)
pipeline = SequentialAgent(name="pipeline", sub_agents=[researcher, writer])

from pydantic import BaseModel

class AnalysisResult(BaseModel):
    summary: str
    key_findings: list[str]
    confidence: float
    recommendations: list[str]

agent = Agent(
    name="analyzer",
    model="gemini-2.5-flash",
    instruction="Analyze the provided data and return structured results.",
    output_schema=AnalysisResult,
    output_key="analysis",  # Stored in state["analysis"]
)

# Install
pip install google-adk

# Set API key
export GOOGLE_API_KEY="your-key"

# Run interactively
adk run my_agent

# Run with web UI
adk web my_agent

import pytest
from google.adk.runners import InMemoryRunner
from google.genai import types

@pytest.mark.asyncio
async def test_agent():
    runner = InMemoryRunner(agent=root_agent, app_name="test")
    session = await runner.session_service.create_session(
        user_id="test_user", app_name="test",
    )
    content = types.Content(
        role="user", parts=[types.Part.from_text(text="Hello")],
    )
    events = []
    async for event in runner.run_async(
        user_id="test_user", session_id=session.id, new_message=content,
    ):
        events.append(event)
    assert "expected" in events[-1].content.parts[0].text.lower()

Single task, one LLM call? → Agent / LlmAgent
Steps must run in order? → SequentialAgent
Steps are independent? → ParallelAgent
Need iteration/refinement? → LoopAgent
Need on-demand delegation? → AgentTool
Remote agent, different service? → RemoteA2aAgent (A2A)
Complex multi-stage? → Compose agent types

Simple function? → Python function with type hints
Need state access? → Add ToolContext parameter
Delegate to another agent? → AgentTool
Remote agent over network? → RemoteA2aAgent (A2A protocol)
External MCP server? → MCPToolset
Database access? → ToolboxToolset
Web search? → google_search (built-in)