pip install google-adk

my_agent/
├── __init__.py
├── agent.py
└── .env          # GOOGLE_API_KEY=xxx or GOOGLE_GENAI_USE_VERTEXAI=TRUE

# Option A: Google AI (simple)
GOOGLE_GENAI_USE_VERTEXAI=FALSE
GOOGLE_API_KEY=your_api_key

# Option B: Vertex AI (enterprise)
GOOGLE_GENAI_USE_VERTEXAI=TRUE
GOOGLE_CLOUD_PROJECT=your_project_id
GOOGLE_CLOUD_LOCATION=us-central1

from google.adk.agents import Agent

root_agent = Agent(
    name="assistant",
    model="gemini-2.0-flash",
    instruction="You are a helpful assistant.",
    tools=[]
)

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

    Args:
        city: City name (e.g., "New York", "London")

    Returns:
        Weather data with temperature and condition
    """
    return {"city": city, "temp": "72F", "condition": "sunny"}

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

# Write state in tools via ToolContext
def save_data(ctx: ToolContext, key: str, value: str) -> str:
    ctx.session.state[key] = value
    return f"Saved {key}"

# Read state in instructions via templating
instruction = "User prefers {user_preference} theme."

# State flows through sub-agents automatically
# Use output_key to capture agent output in state
agent = Agent(name="writer", output_key="draft", ...)

def safety_check(callback_context, llm_request):
    if contains_pii(llm_request):
        return LlmResponse(content="Cannot process PII")
    return None  # Continue normally

agent = Agent(
    name="safe_agent",
    before_model_callback=safety_check,
    # Also: before_agent_callback, after_agent_callback,
    #       after_model_callback, before_tool_callback, after_tool_callback
)

{
  "name": "weather_test",
  "data": [{
    "query": "What's the weather in NYC?",
    "expected_tool_calls": ["get_weather"],
    "reference_answer": "contains temperature"
  }]
}

import streamlit as st
from google.adk.runners import Runner

@st.cache_resource
def get_runner():
    return Runner(agent=root_agent, app_name="my_app")

runner = get_runner()
# Use runner.run() in chat interface

from fastapi import FastAPI
from fastapi.responses import StreamingResponse
from google.adk.runners import Runner
from google.adk.sessions import InMemorySessionService
from google.genai import types

app = FastAPI()
session_service = InMemorySessionService()
runner = Runner(agent=root_agent, app_name="api", session_service=session_service)

@app.post("/chat")
async def chat(request: ChatRequest):
    new_message = types.Content(
        role="user", parts=[types.Part(text=request.message)]
    )
    events = []
    async for event in runner.run_async(
        user_id=request.user_id,
        session_id=request.session_id,
        new_message=new_message,
    ):
        if event.content and event.content.parts:
            for part in event.content.parts:
                if hasattr(part, "text") and part.text:
                    events.append({"type": "content", "text": part.text})
    return {"events": events}

async for event in runner.run_async(user_id=user_id, session_id=session_id, new_message=new_message):
    if event.content and event.content.parts:
        for part in event.content.parts:
            if hasattr(part, "text") and part.text:
                yield part.text

from google.adk.agents import LiveRequestQueue

queue = LiveRequestQueue()
async for event in run_live(agent, queue):
    handle_event(event)

# Expose agent as A2A server
from google.adk.a2a import serve
serve(agent, port=8080)

# Consume remote agent as A2A client
from google.adk.a2a import A2AClient
remote = A2AClient("http://other-agent:8080")