Google Revolutionizes AI Agent Development with the Launch of ADK

From Concept to Production: How Google ADK is Simplifying Intelligent AI Agent Development

Google Revolutionizes AI Agent Development with the Launch of ADK

Yesterday, April 9, 2025, Google officially launched its Agent Development Kit (ADK), a new Python framework that promises to revolutionize AI agent development. As a professional who follows the artificial intelligence market, I consider this launch a significant milestone for those working with AI-based solution development.

What is Google ADK?

ADK is a Python framework designed to simplify the development of applications based on Large Language Models (LLMs). It allows you to create intelligent agents capable of reasoning, planning, using tools, dynamically interacting with users, and collaborating effectively as a team.

Key ADK Differentiators

Multi-agent by design

One of ADK's most impressive aspects is its native multi-agent architecture. You can easily create systems with specialized agents, each responsible for specific tasks, and intelligent delegation between them.

Rich model ecosystem

ADK supports various LLM models, not just Google's. You can implement agents using Gemini, GPT, or Claude, allowing you to choose the most suitable model for each use case.

Pre-built tools

The kit comes with pre-built tools, such as Google Search integration, simplifying the development of agents that need to access web information.

Broad compatibility

One of the great differentiators is compatibility with third-party libraries like LangChain, LlamaIndex, LangGraph, and Crew AI, through an open agent-to-agent protocol.

Flexible orchestration

ADK allows you to create sequential, parallel, or loop workflows, addressing various business needs.

Production-ready

The solution was designed to scale and be deployed in production, leveraging Google's global computing infrastructure.

Getting Started with ADK

The most basic implementation of an agent using ADK is surprisingly simple. Let's look at a practical example in Python:

from google.adk.agents import Agent
from google.adk.tools import Tool

# Defining a simple tool to get weather information
def get_weather(city: str) -> dict:
    """Retrieves weather information for a specific city.
    
    Args:
        city: Name of the city
        
    Returns:
        dict: Weather information with success or error status
    """
    # Simulation of a weather database or API
    weather_data = {
        "Toronto": {"condition": "Partly cloudy", "temperature": "30°C"},
        "São Paulo": {"condition": "Sunny", "temperature": "28°C"},
        "Paris": {"condition": "Rainy", "temperature": "15°C"}
    }
    
    if city in weather_data:
        return {
            "status": "success",
            "weather": weather_data[city]["condition"],
            "temperature": weather_data[city]["temperature"]
        }
    else:
        return {
            "status": "error",
            "message": f"Weather data not available for {city}"
        }

# Creating the agent
weather_agent = Agent(
    name="Weather Agent",
    model="gemini-2.0-flash",  # Specifying the model to use
    description="Provides weather information for specific cities",
    instruction="""You are a helpful weather assistant. 
    Your main goal is to provide current weather reports.
    
    When a user asks about the weather in a specific city, 
    you MUST use the get_weather tool to find this information.
    
    Analyze the tool's response:
    - If the status is "error", politely inform the user about the error message
    - If the status is "success", present the weather report clearly and concisely
    
    Only use the tool when a city is mentioned for a weather request.""",
    tools=[get_weather]  # Providing our tool to the agent
)

# Configuring the session service and runner
from google.adk.sessions import InMemorySessionService
from google.adk.runners import Runner

# Create session service
session_service = InMemorySessionService()
session = session_service.create_session(
    app_name="Weather App",
    user_id="user_123",
    session_id="session_456"
)

# Configure the runner
runner = Runner(
    agent=weather_agent,
    app_name="Weather App",
    session_service=session_service
)

# Example of using the runner
from google.genai.types import Content, Part

# Function to interact with the agent
async def call_agent(query_text: str):
    content = Content(role="user", parts=[Part(text=query_text)])
    async for event in runner.run_async(
        user_id="user_123", 
        session_id="session_456", 
        new_message=content
    ):
        if event.is_final_response():
            return event.content.parts[0].text

# Example usage
# response = await call_agent("What's the weather in São Paulo today?")

Example of creating a multi-agent system

Let's see how to create a system with several specialized agents:

from google.adk.agents import Agent
from google.adk.models.lite_llm import LiteLlm

# Defining specialized agents
greeting_agent = Agent(
    name="Greeting Agent",
    model=LiteLlm(model="anthropic/claude-3-sonnet-20240229"),  # Using Claude for this agent
    description="Handles simple greetings",
    instruction="""You are the greeting agent.
    Your only task is to provide a friendly greeting to the user.
    Do not engage in any other conversation or task.""",
    tools=[say_hello]
)

farewell_agent = Agent(
    name="Farewell Agent",
    model=LiteLlm(model="openai/gpt-4o"),  # Using GPT-4o for this agent
    description="Handles farewells",
    instruction="""You are the farewell agent.
    Your only task is to write a polite goodbye.
    Do not engage in any other conversation or task.""",
    tools=[say_goodbye]
)

# Creating a main agent that coordinates the sub-agents
root_agent = Agent(
    name="Coordinator Agent",
    model="gemini-2.0-flash",
    description="Main agent that coordinates a team of other agents",
    instruction="""You are the main weather agent, coordinating a team of other agents.
    
    Your main task is to provide weather using the get_weather tool,
    handle its status response and report the temperature or send an error message.
    
    Delegation rules for other agents:
    - If the user gives a simple greeting (hello, hi), delegate to the greeting agent
    - If the user gives a farewell (bye, see you), delegate to the farewell agent
    - Handle weather requests yourself using get_weather
    - For other queries, clearly state if you cannot handle them""",
    tools=[get_weather],
    sub_agents=[greeting_agent, farewell_agent]
)

# Configuring the runner to execute the agent
from google.adk.runners import Runner
from google.adk.sessions import InMemorySessionService

# Creating session service
session_service = InMemorySessionService()
session = session_service.create_session("weather_app", "user_123", "session_456")

# Configuring the runner
runner = Runner(
    agent=root_agent,
    app_name="weather_app",
    session_service=session_service
)

Implementing Memory with ADK

One of the most powerful features of ADK is the ability to add memory to your agents. This allows creation of personalized and contextual experiences:

# Implementing an agent with session memory
from google.adk.sessions import InMemorySessionService
from google.adk.tools.tool_context import ToolContext

# Initialize session service with initial state
session_service = InMemorySessionService()
user_id = "user_123"
session_id = "session_789"

# Configure initial state - user preferences
initial_state = {
    "preferences": {
        "temperature_unit": "celsius",
        "previous_searches": []
    }
}

# Create session with initial state
session_service.create_session(
    app_name="weather_app",
    user_id=user_id, 
    session_id=session_id,
    state=initial_state
)

# Tool that uses session context
def stateful_get_weather(city: str, tool_context: ToolContext) -> dict:
    """Gets weather information for a specific city and saves it to the session.
    
    Args:
        city: Name of the city
        tool_context: Context provided by ADK, containing session state
        
    Returns:
        dict: Weather information with status and data
    """
    # Access session state
    state = tool_context.state
    
    # Retrieve preferred temperature unit
    temp_unit = state.get("preferences", {}).get("temperature_unit", "celsius")
    
    # Weather data simulation
    weather_data = {
        "São Paulo": {"condition": "Sunny", "temperature_c": 28, "temperature_f": 82},
    }
    
    if city in weather_data:
        # Choose temperature based on preference
        if temp_unit == "celsius":
            temp = f"{weather_data[city]['temperature_c']}°C"
        else:
            temp = f"{weather_data[city]['temperature_f']}°F"
            
        # Update search history
        if "preferences" in state and "previous_searches" in state["preferences"]:
            state["preferences"]["previous_searches"].append(city)
            # Keep only the 5 most recent searches
            state["preferences"]["previous_searches"] = state["preferences"]["previous_searches"][-5:]
        
        return {
            "status": "success",
            "weather": weather_data[city]["condition"],
            "temperature": temp
        }
    else:
        return {
            "status": "error",
            "error_message": f"Weather data not available for {city}"
        }

# Agent with memory
stateful_agent = Agent(
    name="Weather Agent with Memory",
    model="gemini-2.0-flash",
    description="Provides weather information with customizations based on user preferences",
    instruction="""You are a personalized weather assistant.
    
    Remember user preferences and previous searches to provide personalized responses.
    Use the stateful_get_weather tool when asked about the weather.
    
    If the user has already searched for a city before, mention this in your response.
    If the user wants to change the temperature unit, update the preference.""",
    tools=[stateful_get_weather],
    # Automatically saves the agent's final response to the session state
    output_key="last_weather_report" 
)

Adding Safety Guardrails

An essential functionality for production applications is the ability to implement safety guardrails. ADK offers callbacks that allow controlling agent behavior before potentially problematic requests reach the language model:

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.genai import types
from typing import Optional

# Callback that checks user input before sending to the LLM
def block_keyword_guardrail(
    callback_context: CallbackContext, 
    llm_request: LlmRequest
) -> Optional[LlmResponse]:
    """
    Inspects the user's message for blocked keywords.
    """
    # Get the most recent user message
    last_user_message = ""
    if llm_request.contents:
        for content in reversed(llm_request.contents):
            if content.role == 'user' and content.parts:
                if content.parts[0].text:
                    last_user_message = content.parts[0].text
                    break

    # Check blocked keywords
    blocked_keywords = ["blocked_word1", "blocked_word2"]
    for keyword in blocked_keywords:
        if keyword.lower() in last_user_message.lower():
            # Record in session state
            callback_context.state["guardrail_triggered"] = True
            
            # Block the request and return a custom response
            return LlmResponse(
                content=types.Content(
                    role="model",
                    parts=[types.Part(text=f"This request cannot be processed as it contains disallowed content.")]
                )
            )
    
    # Allow the request to continue
    return None

# Using the callback in the agent
secured_agent = Agent(
    name="Secure Agent",
    model="gemini-2.0-flash",
    description="Agent with safety guardrails",
    instruction="You are a helpful and safe assistant.",
    tools=[get_weather],
    before_model_callback=block_keyword_guardrail
)

Real-time Streaming Features

One of the most impressive functionalities of ADK is support for bidirectional streaming, which enables real-time communication with agents, including text, audio, and video:

from google.adk.agents import Agent
from google.adk.tools import google_search
from google.adk.runners import Runner
from google.adk.agents import LiveRequestQueue
from google.adk.agents.run_config import RunConfig
from google.adk.sessions.in_memory_session_service import InMemorySessionService
from google.genai.types import Part, Content

# Creating a search agent with streaming
search_agent = Agent(
   name="search_agent",
   # For streaming, use the flash model
   model="gemini-2.0-flash-exp",
   description="Agent to answer questions using Google Search.",
   instruction="You are an expert researcher. Always stick to the facts.",
   tools=[google_search]
)

# Configuration for streaming session
APP_NAME = "ADK Streaming App"
session_service = InMemorySessionService()

def start_agent_session(session_id: str):
    """Starts an agent session with streaming"""
    
    # Create session
    session = session_service.create_session(
        app_name=APP_NAME,
        user_id=session_id,
        session_id=session_id,
    )

    # Create Runner
    runner = Runner(
        app_name=APP_NAME,
        agent=search_agent,
        session_service=session_service,
    )

    # Configure response modality as TEXT
    run_config = RunConfig(response_modalities=["TEXT"])

    # Create live request queue for this session
    live_request_queue = LiveRequestQueue()

    # Start streaming session
    live_events = runner.run_live(
        session=session,
        live_request_queue=live_request_queue,
        run_config=run_config,
    )
    return live_events, live_request_queue

This streaming feature is especially useful for:

1. More natural conversational interfaces: The user sees the response being built in real-time, similar to a human conversation

2. Multimodal experiences: Support for audio and video input and output

3. Bidirectional communication: The user can interrupt or send new inputs while the agent is still responding

With ADK Streaming, you can easily integrate your agent into real-time web applications using FastAPI and WebSockets.

Implementing a Web Application with ADK Streaming

ADK also facilitates the creation of web applications with real-time communication. Here's an example using FastAPI:

import os
import json
import asyncio
from fastapi import FastAPI, WebSocket
from fastapi.staticfiles import StaticFiles
from fastapi.responses import FileResponse

from google.genai.types import Part, Content
from google.adk.runners import Runner
from google.adk.agents import LiveRequestQueue
from google.adk.agents.run_config import RunConfig
from google.adk.sessions.in_memory_session_service import InMemorySessionService

from my_package.agents import root_agent  # Your main agent

# Application configuration
APP_NAME = "ADK Streaming Example"
session_service = InMemorySessionService()

# Function to start agent session with streaming
def start_agent_session(session_id: str):
    # Create session
    session = session_service.create_session(
        app_name=APP_NAME,
        user_id=session_id,
        session_id=session_id,
    )

    # Create Runner
    runner = Runner(
        app_name=APP_NAME,
        agent=root_agent,
        session_service=session_service,
    )

    # Configure response modality as TEXT
    run_config = RunConfig(response_modalities=["TEXT"])

    # Create live request queue for this session
    live_request_queue = LiveRequestQueue()

    # Start streaming session
    live_events = runner.run_live(
        session=session,
        live_request_queue=live_request_queue,
        run_config=run_config,
    )
    return live_events, live_request_queue

# Agent to client communication
async def agent_to_client(websocket, live_events):
    async for event in live_events:
        # Turn completion event
        if event.turn_complete:
            await websocket.send_text(json.dumps({"turn_complete": True}))
            
        # Get partial text from agent response
        part = event.content and event.content.parts and event.content.parts[0]
        if not part or not event.partial:
            continue
            
        # Send text to client
        text = part.text
        if text:
            await websocket.send_text(json.dumps({"message": text}))
            await asyncio.sleep(0)

# Client to agent communication
async def client_to_agent(websocket, live_request_queue):
    while True:
        text = await websocket.receive_text()
        content = Content(role="user", parts=[Part.from_text(text=text)])
        live_request_queue.send_content(content=content)
        await asyncio.sleep(0)

# FastAPI application
app = FastAPI()

@app.websocket("/ws/{session_id}")
async def websocket_endpoint(websocket: WebSocket, session_id: int):
    # Accept client connection
    await websocket.accept()
    
    # Start agent session
    session_id = str(session_id)
    live_events, live_request_queue = start_agent_session(session_id)
    
    # Start bidirectional communication tasks
    agent_to_client_task = asyncio.create_task(
        agent_to_client(websocket, live_events)
    )
    client_to_agent_task = asyncio.create_task(
        client_to_agent(websocket, live_request_queue)
    )
    
    # Run both tasks concurrently
    await asyncio.gather(agent_to_client_task, client_to_agent_task)

This code implements a web application that allows:

• Real-time bidirectional communication between the user and the agent

• Streaming of partial responses (the user sees the response being built)

• Support for multiple simultaneous sessions

• Scalability for adding audio and video support

Testing and Deploying Your Agent

One of ADK's strengths is support for local testing and simplified deployment. Let's see how to test your agent locally before putting it into production:

# Recommended directory structure
adk_project/
  |- my_agent/
     |- __init__.py
     |- .env  # For storing API keys
     |- agent.py

Testing Locally with the Integrated API Server

ADK offers a simple way to start a local FastAPI server to test your agent:

# In the project root directory
adk api_server

This starts a local server at http://0.0.0.0:8000 where you can interact with your agent using simple HTTP calls:

# Create a new session for the agent
curl -X POST http://0.0.0.0:8000/apps/my_agent/users/user_123/sessions/session_456 \
  -H "Content-Type: application/json" \
  -d '{"state": {"temperature_preference": "celsius"}}'

# Send a query to the agent
curl -X POST http://0.0.0.0:8000/run \
-H "Content-Type: application/json" \
-d '{
  "app_name": "my_agent",
  "user_id": "user_123",
  "session_id": "session_456",
  "new_message": {
    "role": "user",
    "parts": [{
      "text": "What's the weather in São Paulo today?"
    }]
  }
}'

ADK also supports stream communication through the /run_sse endpoint:

curl -X POST http://0.0.0.0:8000/run_sse \
-H "Content-Type: application/json" \
-d '{
  "app_name": "my_agent",
  "user_id": "user_123",
  "session_id": "session_456",
  "new_message": {
    "role": "user",
    "parts": [{
      "text": "What's the weather in São Paulo today?"
    }]
  },
  "streaming": true
}'

Deployment Options

After testing your agent locally, you have several deployment options:

1. Agent Engine on Vertex AI: The easiest way to deploy ADK agents on Google Cloud, with automatic scaling and simplified management.

2. Cloud Run: You can customize how the container scales and operates for greater control over infrastructure.

Both options offer:

• Automatic scaling to handle variable loads

• Integration with the Google Cloud ecosystem

• Integrated monitoring and logging

• High availability

Why Adopt ADK Now?

Over the past year, various companies have launched their own frameworks for agent development. Google ADK's differentiator is that it has learned from the limitations of these previous systems, offering a more complete and interoperable solution.

Google's agent-to-agent protocol allows you to integrate existing systems based on LangChain, LlamaIndex, or Crew AI, preserving your previous investments while evolving your architecture.

# Example of LangChain integration
from langchain.tools import Tool as LangChainTool
from google.adk.tools import Tool as AdkTool

# Existing LangChain tool
langchain_search_tool = LangChainTool(
    name="web_search",
    description="Searches the web for information",
    func=lambda query: "Search results for: " + query
)

# Adapting to ADK
adk_search_tool = AdkTool.from_langchain(langchain_search_tool)

# Using in an ADK agent
search_agent = Agent(
    name="Search Agent",
    model="gemini-2.0-flash",
    description="Performs web searches",
    instruction="Use the search tool to find information requested by the user.",
    tools=[adk_search_tool]
)

If you're considering entering the AI agent development field or have already worked with it, now is the ideal time to explore Google ADK. The open architecture and integration with existing tools allow for a gradual migration without the need to abandon previous investments.

Conclusion

Google ADK represents a significant advancement in democratizing AI agent development. It solves many of the problems faced by previous frameworks, offering:

1. Native multi-agent architecture: Allows intuitive creation of complex and modular systems.

2. Compatibility with multiple models: Flexibility to choose between Gemini, GPT, Claude, or any other LLM.

3. Consistent and well-documented API: This significantly reduces the code to create powerful applications.

4. Integrated advanced features: Bidirectional streaming, state management, safety guardrails, and more.

5. Interoperability: It works with existing frameworks like LangChain, Llama Index, and others.

With Google ADK, we can create more complex and valuable agent systems with less code and implementation complexity. If you're interested in building advanced solutions based on generative AI, this is the perfect time to explore ADK.

I'm particularly excited about the possibilities for enterprise applications, personalized assistants, and intelligent automation. If you're also exploring this path, share your impressions about this new technology in the comments!

#GoogleADK #ArtificialIntelligence #AgentDevelopment #AI #TechNews #Innovation