Google ADK: Modular AI Agent Development from Scratch

import os
import asyncio
from dotenv import load_dotenv
import logging

# Load environment variables from .env file
load_dotenv()

GOOGLE_API_KEY = os.getenv("GOOGLE_API_KEY")

# Configure ADK to use the API key directly
os.environ["GOOGLE_GENAI_USE_VERTEXAI"] = "False"

# Define the default model to use across the notebook
MODEL = "gemini-2.5-flash"

# Suppress verbose SDK warning about non-text parts when tools/structured outputs are present
class _SuppressNonTextParts(logging.Filter):
    def filter(self, record: logging.LogRecord) -> bool:
        try:
            msg = record.getMessage()
        except Exception:
            msg = str(record.msg)
        return "non-text parts in the response" not in msg

for _name in ("google", "google.genai"):
    _logger = logging.getLogger(_name)
    _logger.addFilter(_SuppressNonTextParts())

if not GOOGLE_API_KEY:
    print("API key not found. Please create a .env file and add your GOOGLE_API_KEY.")
else:
    print("Successfully loaded API key.")
    print(f"Using model: {MODEL}")

from google.adk.agents import Agent
from typing import Optional

# --- Define Tools ---
def get_weather(city: str) -> dict:
    """Retrieves the current weather report for a specified city."""
    print(f"--- Tool: get_weather called for city: {city} ---")
    mock_db = {
        "new york": {"report": "The weather in New York is sunny with a temperature of 25°C."},
        "london": {"report": "It's cloudy in London with a temperature of 15°C."},
    }
    return mock_db.get(city.lower(), {"report": f"Sorry, I don't have weather for {city}."})


def say_hello(name: Optional[str] = None) -> str:
    """Provides a simple friendly greeting."""
    print(f"--- Tool: say_hello called ---")
    return f"Hello, {name}!" if name else "Hello there!"

# --- Define Specialist Agents ---
weather_agent = Agent(
    name="WeatherAgent",
    model=MODEL,
    description=(
        "Provides weather information for specific cities. Use this for any query that mentions weather, "
        "forecast, temperature, rain, sun, or a city/location."
    ),
    instruction=(
        "You are a helpful weather assistant. Always use the get_weather tool to obtain information. "
        "Respond concisely with the weather report."
    ),
    tools=[get_weather],
)

greeting_agent = Agent(
    name="GreetingAgent",
    model=MODEL,
    description=(
        "Handles only pure greetings such as 'hi', 'hello', or 'hey' with no other request. "
        "Do not handle any other topics (e.g., weather)."
    ),
    instruction=(
        "You are the Greeting Agent. Respond ONLY to standalone greetings (hi/hello/hey) with a short greeting. "
        "If the user asks about anything else (e.g., weather, cities, temperatures, or questions), do NOT answer; "
        "leave it to other agents."
    ),
    tools=[say_hello],
)

print("✅ Specialist agents and tools defined.")

# The root agent orchestrates the team
root_agent = Agent(
    name="RootAgent",
    model=MODEL,
    instruction=(
        "You are the coordinator of an agent team. Route each user request to the most appropriate sub-agent.\n"
        "Routing rules:\n"
        "- If the user mentions weather, forecast, temperature, a city/location, or asks about conditions, delegate to WeatherAgent.\n"
        "- If the user gives a standalone greeting (hi/hello/hey) with no other request, delegate to GreetingAgent.\n"
        "- Do NOT answer yourself; always delegate to a sub-agent.\n"
        "- Never treat a weather-related question as a greeting."
    ),
    # Key step: Link the sub-agents to enable delegation
    sub_agents=[weather_agent, greeting_agent]
)
print(f"✅ Root Agent '{root_agent.name}' created with sub-agents: {[sa.name for sa in root_agent.sub_agents]}")

from google.adk.runners import Runner
from google.adk.sessions import InMemorySessionService
from google.genai import types
import sys
import io
import contextlib

runner = Runner(
    agent=root_agent,
    session_service=InMemorySessionService(),
    app_name="weather_tutorial_app"
)

class _FilterStream(io.TextIOBase):
    def __init__(self, underlying, substrings):
        self._underlying = underlying
        self._substrings = tuple(substrings)

    def write(self, s):
        try:
            text = str(s)
        except Exception:
            text = s
        if any(sub in text for sub in self._substrings):
            return len(text)
        return self._underlying.write(text)

    def flush(self):
        return self._underlying.flush()

async def call_agent(query: str, session_id: str):
    """Helper function to run an asynchronous query against the agent."""
    print(f"\n>>> User: {query}")

    # Filter noisy stderr warnings about non-text parts when tools are used
    filter_stream = _FilterStream(sys.stderr, ["non-text parts in the response"])
    with contextlib.redirect_stderr(filter_stream):
        # Use run_async to get events and collect only the final final-response
        final_text = None
        async for event in runner.run_async(
            user_id="user_123",
            session_id=session_id,
            new_message=types.Content(role='user', parts=[types.Part(text=query)])
        ):
            if event.is_final_response() and event.content:
                # capture text but don't print yet to avoid duplicates
                for part in event.content.parts:
                    if getattr(part, "text", None):
                        final_text = part.text
                        break
        if final_text:
            print(f"<<< Agent: {final_text}")

# Let's run a multi-turn conversation!
async def main():
    session_id = "session_abc"
    user_id = "user_123"

    # Ensure the session exists before running; required by InMemorySessionService
    await runner.session_service.create_session(
        app_name="weather_tutorial_app",
        user_id=user_id,
        session_id=session_id,
    )

    print("--- Starting conversation ---")

    # The root agent should delegate this to the GreetingAgent
    await call_agent("Hello there", session_id)

    # The root agent should delegate this to the WeatherAgent
    await call_agent("What is the weather in New York?", session_id)

# In a Jupyter Notebook, you can run the top-level async function with await
await main()