Exploring Google’s Agent Development Kit (ADK)

Orchestrating Intelligent Agentic Systems — from Simple Tools to Complex Workflows

This article has an accompanying GitHub Repository containing runnable samples:

In recent years, the development of autonomous agents — software entities capable of reasoning, planning, and taking actions on behalf of users — has moved from research labs into real-world applications. These AI agents are rapidly becoming central to building intelligent systems, whether through task automation, information retrieval, or multi-step problem solving.

To support this growing demand, Google has introduced the , an open-source framework designed to make it easier to build, orchestrate, and evaluate agents powered by . ADK offers developers a structured way to compose agents that can interact with tools, communicate with each other, and reason about their actions — all while integrating with Google’s own Gemini models or other LLMs.

Unlike many frameworks that offer minimal guidance beyond basic function calls, ADK takes a more holistic approach. It provides abstractions for agent behavior, decision-making loops, and tool usage, while also encouraging responsible development practices. This makes it a compelling option for researchers exploring multi-agent systems and engineers looking to build production-ready AI workflows.

In this post, we’ll take a deep dive into the ADK: its architecture, the types of agents it supports, how to define and use tools, and how to evaluate your agents in realistic scenarios.

Agents in ADK: Types and Implementation

In the Agent Development Kit, an agent is the fundamental building block. It’s a self-contained unit that can take in input, make decisions (sometimes with the help of an LLM), and call tools to act on behalf of a user or system. What sets ADK apart is its support for multiple types of agents, each optimized for different workflows.

Let’s explore the key agent types in ADK.

1. LLMAgent

An is the most flexible and dynamic agent type. It uses a large language model (like Gemini) to decide which tool to call and how to call it based on natural language instructions and intermediate results.

This makes it ideal for:

• Open-ended tasks
• Natural language user inputs
• Complex workflows with unclear structure

How It Works:

• You provide a prompt, a set of tools, and the agent uses the LLM to decide how to proceed.
• Internally, the LLM receives a “scratchpad” of previous actions and responses, enabling iterative reasoning.

Code Example:

from google.adk.agents import LlmAgent
from google.adk.runners import Runner
from google.adk.sessions import InMemorySessionService
from google.genai import types

# Define a simple tool
def get_weather(city: str) -> str:
    """Gets the current weather for a city.
    Args:
        city: The name of the city to get weather for.
    Returns:
        A string with the weather description.
    """
    return f"The weather in {city} is sunny."
# Create the agent
agent = LlmAgent(
    name="weather_agent",
    model="gemini-2.0-flash",
    tools=[get_weather],
    description="A helpful assistant that can check weather."
)
# Set up runner and session for execution
APP_NAME = "weather_app"
USER_ID = "user_123"
SESSION_ID = "session_456"
# Create session service and session
session_service = InMemorySessionService()
session = session_service.create_session(
    app_name=APP_NAME,
    user_id=USER_ID,
    session_id=SESSION_ID
)
# Create runner
runner = Runner(
    agent=agent,
    app_name=APP_NAME,
    session_service=session_service
)
# Run the agent using the runner
def run_query(query):
    # Create content from user query
    content = types.Content(
        role="user",
        parts=[types.Part(text=query)]
    )
    # Run the agent with the runner
    events = runner.run(
        user_id=USER_ID,
        session_id=SESSION_ID,
        new_message=content
    )
    # Process events to get the final response
    for event in events:
        if event.is_final_response():
            return event.content.parts[0].text
    return "No response received."
# Example of usage
response = run_query("What's the weather like in Berlin?")
print(response)

2. SequentialAgent

A is a rule-based agent that executes tools in a specific order. There's no decision-making involved — it simply runs step-by-step through a predefined sequence.

Best used for:

• Rigid data pipelines
• ETL tasks (extract, transform, load)
• Anything that follows a known path

Code Example:

from google.adk.agents import SequentialAgent, LlmAgent
from google.adk.runners import Runner
from google.adk.sessions import InMemorySessionService
from google.genai import types

# Define tools as functions
def extract_data(input_text: str) -> str:
    """Extracts raw data from input.
    Args:
        input_text: The text to extract data from.
    Returns:
        The extracted raw data.
    """
    return input_text.upper()

def clean_data(data: str) -> str:
    """Cleans the provided data.
    Args:
        data: The data to clean.
    Returns:
        The cleaned data.
    """
    return data.strip()

# Create individual agents for each step
extract_agent = LlmAgent(
    name="extract_agent",
    model="gemini-2.0-flash",
    tools=[extract_data],
    instruction=(
        "You are a data extraction agent. "
        "When given the user message as `input_text`, "
        "call the Python function `extract_data(input_text)` and return *only* its output."
    ),
    description="Extracts data from input",
    output_key="raw_data"
)
clean_agent = LlmAgent(
    name="clean_agent",
    model="gemini-2.0-flash",
    tools=[clean_data],
    instruction=(
        "You are a data cleaning agent. "
        "Take the extracted data in `raw_data`, "
        "call the Python function `clean_data(raw_data)`, and return *only* the cleaned data."
    ),
    description="Cleans extracted data"
)
# Create sequential agent with the proper sub-agents
sequential_agent = SequentialAgent(
    name="sequential_pipeline",
    sub_agents=[extract_agent, clean_agent],
    description="Runs extract_agent then clean_agent, in order."
)
# Set up runner and session for execution
APP_NAME = "sequential_app"
USER_ID = "user_123"
SESSION_ID = "session_456"
# Create session service and session
session_service = InMemorySessionService()
session = session_service.create_session(
    app_name=APP_NAME,
    user_id=USER_ID,
    session_id=SESSION_ID
)
# Create runner
runner = Runner(
    agent=sequential_agent,
    app_name=APP_NAME,
    session_service=session_service
)

# Run the sequential agent using the runner
def run_sequential_agent(input_text):
    # Create content from user input
    content = types.Content(
        role="user",
        parts=[types.Part(text=input_text)]
    )
    # Run the agent with the runner
    final_response = "No response received."
    for event in runner.run(
            user_id=USER_ID,
            session_id=SESSION_ID,
            new_message=content
    ):
        if event.is_final_response():
            final_response = event.content.parts[0].text
    return final_response

# Example usage
result = run_sequential_agent("start")
print(result)

3. ParallelAgent

A runs multiple tools concurrently through async I/O and returns their results as a batch. While this provides a performance boost for I/O-bound operations, note that this is not necessarily multi-threaded parallelism. This is useful when:

• Tasks are independent of each other
• You want to speed up workflows
• You’re aggregating results (e.g., fetching from multiple APIs)

Code Example:

from google.adk.agents import ParallelAgent, LlmAgent
from google.adk.runners import Runner
from google.adk.sessions import InMemorySessionService
from google.genai import types

# Define some example tools
def get_weather(city: str) -> str:
    """Gets the current weather for a city."""
    return f"The weather in {city} is sunny."

def get_news(topic: str) -> str:
    """Gets the latest news on a topic."""
    return f"Latest news about {topic}: Everything is great!"

# Create individual agents for different tasks
weather_agent = LlmAgent(
    name="weather_agent",
    model="gemini-2.0-flash",
    tools=[get_weather],
    instruction=(
        "Extract the city name from the user query, call get_weather(city), "
        "and return only that result."
    ),
    description="Provides weather information",
    output_key = "weather_info"
)
news_agent = LlmAgent(
    name="news_agent",
    model="gemini-2.0-flash",
    tools=[get_news],
    instruction=(
        "Extract the topic from the user query, call get_news(topic), "
        "and return only that result."
    ),
    description="Provides news updates",
    output_key="news_info"
)
# Create a parallel agent with proper agents as sub-agents
parallel_agent = ParallelAgent(
    name="parallel_fetcher",
    sub_agents=[weather_agent, news_agent],
    description="Fetch weather & news at the same time"
)
# Set up runner and session for execution
APP_NAME = "parallel_app"
USER_ID = "user_123"
SESSION_ID = "session_456"
# Create session service and session
session_service = InMemorySessionService()
session = session_service.create_session(
    app_name=APP_NAME,
    user_id=USER_ID,
    session_id=SESSION_ID
)
# Create runner
runner = Runner(
    agent=parallel_agent,
    app_name=APP_NAME,
    session_service=session_service
)

# Run the parallel agent using the runner
def run_parallel_agent(query):
    # Create content from user query
    content = types.Content(
        role="user",
        parts=[types.Part(text=query)]
    )
    replies = []
    # Run the agent with the runner
    for event in runner.run(
            user_id=USER_ID,
            session_id=SESSION_ID,
            new_message=content
    ):
        if event.content:
            replies.append(event.content.parts[0].text)
    return replies
# Example usage
result = run_parallel_agent("Berlin")
print(result)

4. LoopAgent

A repeats a specific tool or agent until a condition is met. It's essentially a control structure — like a while loop — that enables iterative refinement or repeated querying.

Great for:

• Searching and summarizing
• Multi-step reasoning until success
• Self-checking agents

Code Example:

from google.adk.agents import LoopAgent, LlmAgent, BaseAgent
from google.adk.runners import Runner
from google.adk.sessions import InMemorySessionService
from google.genai import types
from google.adk.events import Event, EventActions

# Define tools
def guess_number(input_text: str) -> str:
    """Makes a guess at the target number.
    Args:
        input_text: Context for the guess.
    Returns:
        A string containing the guessed number.
    """
    # This is just a mockup - would have actual logic in a real implementation
    return "Is it 42?"

# Create an agent for guessing
guess_agent = LlmAgent(
    name="guesser",
    model="gemini-2.0-flash",
    description="Makes a guess at the target number.",
    instruction=(
        "You are a number-guessing agent. On each turn, "
        "call `guess_number(input_text)` and return *only* its output."
    ),
    tools=[guess_number],
    output_key="last_response"
)

# Create a custom checker agent that determines when to stop looping
class CheckerAgent(BaseAgent):
    """Agent that checks if the guessed number is correct."""
    def __init__(self, name: str):
        super().__init__(name=name)
    async def _run_async_impl(self, context):
        # pull the last guess out of state
        last = context.session.state.get("last_response", "")
        # keep looping until we actually saw "42"
        found = "42" in last
        # "continue" vs "stop" is your protocol
        verdict = "stop" if found else "continue"
        # ALWAYS supply an EventActions instance (cannot be None)
        actions = EventActions(escalate=found)
        yield Event(
            author=self.name,
            content=types.Content(
                role="assistant",
                parts=[types.Part(text=verdict)]
            ),
            actions=actions
        )

# Create the checker agent
checker_agent = CheckerAgent(name="checker")
# Create loop agent with proper configuration
loop_agent = LoopAgent(
    name="guessing_loop",
    sub_agents=[guess_agent, checker_agent],
    max_iterations=5,
    description="Repeatedly guesses a number until correct or max iterations reached",
)
# Set up runner and session for execution
APP_NAME = "loop_app"
USER_ID = "user_123"
SESSION_ID = "session_456"
# Create session service and session
session_service = InMemorySessionService()
session = session_service.create_session(
    app_name=APP_NAME,
    user_id=USER_ID,
    session_id=SESSION_ID
)
# Create runner
runner = Runner(
    agent=loop_agent,
    app_name=APP_NAME,
    session_service=session_service
)

# Run the loop agent using the runner
def run_loop_agent(query):
    # Create content from user query
    content = types.Content(
        role="user",
        parts=[types.Part(text=query)]
    )
    # Run the agent with the runner
    for event in runner.run(
            user_id=USER_ID,
            session_id=SESSION_ID,
            new_message=content
    ):
        if event.is_final_response():
            return event.content.parts[0].text
    return "No response received."

# Example usage
result = run_loop_agent("Start guessing")
print(result)

5. Nested Agents (Agent-as-a-Tool)

In ADK, you can treat an entire agent as a tool. This means one agent can call another agent as part of its workflow. This is particularly powerful for:

• Decomposing tasks into sub-agents
• Creating reusable building blocks
• Building hierarchical systems

Code Example:

from google.adk.agents import LlmAgent
from google.adk.tools.agent_tool import AgentTool
from google.adk.runners import Runner
from google.adk.sessions import InMemorySessionService
from google.genai import types

# Assuming we have a pre-existing agent
sub_agent = LlmAgent(
    name="specialized_agent",
    model="gemini-2.0-flash",
    description="A specialized agent with specific capabilities",
    instruction=(
        "You are the specialized agent. Take the user's request string "
        "and return exactly: "Specialized result: <their request>"."
    )
)
# Create an agent tool from the sub-agent
nested_tool = AgentTool(agent=sub_agent)
# Use this tool in another agent
super_agent = LlmAgent(
    name="supervisor",
    model="gemini-2.0-flash",
    tools=[nested_tool],
    output_key="delegated_response",
    description="Delegates the user's request to a specialist and returns the result.",
    instruction=(
        "You are the supervisor. When the user gives you a request, "
        "call `specialized_tool(request)` and return *only* what that tool returns."
    ))
# Set up runner and session for execution
APP_NAME = "delegation_app"
USER_ID = "user_123"
SESSION_ID = "session_456"
# Create session service and session
session_service = InMemorySessionService()
session = session_service.create_session(
    app_name=APP_NAME,
    user_id=USER_ID,
    session_id=SESSION_ID
)
# Create runner
runner = Runner(
    agent=super_agent,
    app_name=APP_NAME,
    session_service=session_service
)
def run_supervisor(query: str) -> str:
    msg = types.Content(role="user", parts=[types.Part(text=query)])
    for event in runner.run(user_id=USER_ID, session_id=SESSION_ID, new_message=msg):
        if event.is_final_response():
            return event.content.parts[0].text
    return "No response received."

# Example usage
response = run_supervisor("Please transform this text")
print(response)

Tools in ADK: Building Blocks of Agent Behavior

Agents may make decisions and plan workflows, but they can’t do anything useful without tools. In ADK, a tool is a wrapper around a function or capability that the agent can call to take real-world actions — like retrieving information, processing data, sending messages, or even invoking another agent.

Think of tools as the “hands” of your agent: they do the actual work, while the agent decides what to do and when.

What Is a Tool in ADK?

A tool in ADK is a standardized Python object that contains:

• A name: how the agent refers to it
• A function: the callable logic that performs the action
• An optional description: used by LLM agents to decide which tool to use
• Optionally: input/output schemas to validate inputs or improve prompting

This abstraction makes it possible for agents to use tools interchangeably, even if they were built by different developers or perform wildly different tasks.

1. Creating a Basic Tool

Let’s start with the simplest use case: defining a Python function to use as a tool.

from google.adk.agents import LlmAgent
from google.adk.runners import Runner
from google.adk.sessions import InMemorySessionService
from google.genai import types

def greet_user(name: str) -> str:
    """Greets the user by name.
    Args:
        name: The name of the user to greet.
    Returns:
        A greeting message.
    """
    return f"Hello, {name}!"
# Create the agent with the function as a tool
greeting_agent = LlmAgent(
    name="greeter",
    model="gemini-2.0-flash",
    tools=[greet_user],
    description="Agent that can greet users"
)
# Set up runner and session
APP_NAME = "greeting_app"
USER_ID = "user_123"
SESSION_ID = "session_456"
# Create session service and session
session_service = InMemorySessionService()
session = session_service.create_session(
    app_name=APP_NAME,
    user_id=USER_ID,
    session_id=SESSION_ID
)
# Create runner
runner = Runner(
    agent=greeting_agent,
    app_name=APP_NAME,
    session_service=session_service
)
# Run the agent
def run_greeting_agent(query):
    content = types.Content(
        role="user",
        parts=[types.Part(text=query)]
    )
    for event in runner.run(
        user_id=USER_ID,
        session_id=SESSION_ID,
        new_message=content
    ):
        if event.is_final_response():
            return event.content.parts[0].text
    return "No response received."
response = run_greeting_agent("Say hello to Alice")
print(response)

2. Adding Tool Context Access

For more advanced scenarios, ADK allows tools to access contextual information using the special tool_context parameter.

from google.adk.agents import LlmAgent
from google.adk.runners import Runner
from google.adk.sessions import InMemorySessionService
from google.adk.tools import ToolContext
from google.genai import types

# Tool with context access
def process_document(document_name: str, analysis_query: str, tool_context: ToolContext) -> dict:
    """Analyzes a document using context from memory.
    Args:
        document_name: Name of the document to analyze.
        analysis_query: The specific analysis to perform.
        tool_context: Access to state and other context variables.
    Returns:
        A dictionary with analysis results or error information.
    """
    # Access session state
    previous_queries = tool_context.state.get("previous_queries", [])
    # Update state with the new query
    tool_context.actions.state_delta = {
        "previous_queries": previous_queries + [analysis_query]
    }
    # Actual document processing would go here
    return {
        "status": "success",
        "analysis": f"Analysis of '{document_name}' regarding '{analysis_query}'"
    }

# Create the agent with the tool
document_agent = LlmAgent(
    name="document_analyzer",
    model="gemini-2.0-flash",
    output_key="analysis_result",
    tools=[process_document],
    description="Analyzes a document and records each query in history.",
    instruction=(
        "You are a document analysis agent. When the user asks "
        "'Analyze <document_name> for <analysis_query>', extract both parts, "
        "call `process_document(document_name, analysis_query, tool_context)`, "
        "and return *only* the resulting JSON dictionary."
    )
)
# Set up runner and session
APP_NAME = "document_app"
USER_ID = "user_123"
SESSION_ID = "session_456"
# Create session service and session
session_service = InMemorySessionService()
session = session_service.create_session(
    app_name=APP_NAME,
    user_id=USER_ID,
    session_id=SESSION_ID
)
# Create runner
runner = Runner(
    agent=document_agent,
    app_name=APP_NAME,
    session_service=session_service
)

# Run the agent
def run_document_agent(query):
    content = types.Content(
        role="user",
        parts=[types.Part(text=query)]
    )
    for event in runner.run(
            user_id=USER_ID,
            session_id=SESSION_ID,
            new_message=content
    ):
        if event.is_final_response():
            return event.content.parts[0].text
    return "No response received."

# Example usage
response = run_document_agent("Analyze report.pdf for sales trends")
print(response)

3. Tool Composition: Tools That Use Other Tools

You can create tools that call other tools, enabling more complex behaviors:

from google.adk.agents import LlmAgent
from google.adk.runners import Runner
from google.adk.sessions import InMemorySessionService
from google.genai import types

# Basic tool
def get_data(source: str) -> str:
    """Gets raw data from a specified source.
    Args:
        source: The name or URL of the data source.
    Returns:
        The raw data as a string.
    """
    return f"Raw data from {source}: 42, 17, 23, 8, 15"

# Tool that uses the first tool
def analyze_data(source: str) -> str:
    """Analyzes data from a specified source.
    Args:
        source: The name or URL of the data source.
    Returns:
        Analysis of the data.
    """
    # Get the data first
    raw_data = get_data(source)
    # Parse and analyze it
    numbers = [int(n.strip()) for n in raw_data.split(":", 1)[1].split(",")]
    total = sum(numbers)
    average = total / len(numbers)
    return f"Analysis of {source}:\n- Total: {total}\n- Average: {average:.2f}"

# Create the agent with both tools
data_agent = LlmAgent(
    name="data_analyzer",
    model="gemini-2.0-flash",
    tools=[analyze_data],
    output_key="analysis_summary",
    description="Fetches raw data and returns its analysis.",
    instruction=(
        "You are the data analyzer. When the user says "
        "'Analyze data from <source>', extract the <source> string, "
        "call `analyze_data(source)`, and return *only* its output."
    )
)
# Set up runner and session
APP_NAME = "data_app"
USER_ID = "user_123"
SESSION_ID = "session_456"
# Create session service and session
session_service = InMemorySessionService()
session = session_service.create_session(
    app_name=APP_NAME,
    user_id=USER_ID,
    session_id=SESSION_ID
)
# Create runner
runner = Runner(
    agent=data_agent,
    app_name=APP_NAME,
    session_service=session_service
)

# Run the agent
def run_data_agent(query):
    content = types.Content(
        role="user",
        parts=[types.Part(text=query)]
    )
    for event in runner.run(
            user_id=USER_ID,
            session_id=SESSION_ID,
            new_message=content
    ):
        if event.is_final_response():
            return event.content.parts[0].text
    return "No response received."

# Example usage
response = run_data_agent("Analyze data from database_alpha")
print(response)

Creating and Using Tools

When using a standard Python function as an ADK Tool, how you define it significantly impacts the agent’s ability to use it correctly. The LLM relies heavily on the function’s name, parameters, type hints, and docstring to understand its purpose and generate the correct call.

In ADK, you can directly pass Python functions to the agent’s tools list, and the framework will automatically wrap them as Function Tools:

from google.adk.agents import LlmAgent, SequentialAgent
from google.adk.runners import Runner
from google.adk.sessions import InMemorySessionService
from google.genai import types

# Define tool functions
def extract(input_text: str) -> str:
    """Extracts data from input."""
    return "Extracted: " + input_text

def clean(data: str) -> str:
    """Cleans and processes the data."""
    return data.strip()

def summarize(data: str) -> str:
    """Summarizes the provided data."""
    return f"Summary of: {data}"

# Create individual agents with direct function references
extract_agent = LlmAgent(
    name="extractor",
    model="gemini-2.0-flash",
    tools=[extract],
    output_key="extracted_data",
    description="Extracts raw data from the user's input.",
    instruction=(
        "You are the extractor. Given the user's message as `input_text`, "
        "call `extract(input_text)` and return *only* its output."
    )
)
clean_agent = LlmAgent(
    name="cleaner",
    model="gemini-2.0-flash",
    tools=[clean],
    output_key="cleaned_data",
    description="Cleans the extracted data.",
    instruction=(
        "You are the cleaner. Take the extracted data in `extracted_data`, "
        "call `clean(extracted_data)`, and return *only* the cleaned result."
    )
)
summarize_agent = LlmAgent(
    name="summarizer",
    model="gemini-2.0-flash",
    tools=[summarize],
    output_key="final_summary",
    description="Summarizes the cleaned data.",
    instruction=(
        "You are the summarizer. Take the cleaned data in `cleaned_data`, "
        "call `summarize(cleaned_data)`, and return *only* the summary."
    )
)
# Create a sequential pipeline
pipeline = SequentialAgent(
    name="data_pipeline",
    sub_agents=[extract_agent, clean_agent, summarize_agent],
    description="Processes data in three steps: extract, clean, summarize"
)
# For ADK tools compatibility, the root agent must be named `root_agent`
root_agent = pipeline
# Set up runner and session for execution
APP_NAME = "pipeline_app"
USER_ID = "user_123"
SESSION_ID = "session_456"
# Create session service and session
session_service = InMemorySessionService()
session = session_service.create_session(
    app_name=APP_NAME,
    user_id=USER_ID,
    session_id=SESSION_ID
)
# Create runner
runner = Runner(
    agent=root_agent,
    app_name=APP_NAME,
    session_service=session_service
)

# Function to run the pipeline
def run_pipeline(input_text):
    # Create content from user input
    content = types.Content(
        role="user",
        parts=[types.Part(text=input_text)]
    )
    # Run the agent with the runner
    final_response = "No response received."
    for event in runner.run(
            user_id=USER_ID,
            session_id=SESSION_ID,
            new_message=content
    ):
        # Check for intermediate state changes (debugging)
        if event.actions and event.actions.state_delta:
            print(f"State update: {event.actions.state_delta}")
        # Get final response
        if event.is_final_response():
            final_response = event.content.parts[0].text
    return final_response

# Example usage
result = run_pipeline("This is some sample input data.")
print(result)

Orchestrating Agents: Composing Complex Workflows

So far, we’ve looked at agents and tools in isolation. But real-world problems rarely fit into neat one-step interactions. Tasks often involve multiple decisions, fallback strategies, loops, and collaboration between different agents.

ADK offers several orchestration patterns that let you move beyond single-agent logic. Whether you’re chaining steps linearly, running processes in parallel, or dynamically delegating subtasks, orchestration in ADK is what gives your agent systems depth and flexibility.

Why Orchestration Matters

Without orchestration, an agent is like a solo script — great for single-purpose tasks but limited in scope. With orchestration, you can:

• Break down complex goals into modular components
• Delegate subtasks to specialized agents
• Run parts of your system in parallel for performance
• Introduce retry mechanisms, loops, or condition-based routing
• Build resilient, testable, maintainable systems

Key Patterns in ADK Orchestration

Let’s walk through the most common orchestration strategies supported by ADK — with implementation examples.

1. Sequential Composition

You can chain multiple tools or agents in a predefined order using a SequentialAgent. Think of this like a pipeline: data flows from one step to the next.

Example: Document Processing Pipeline

from google.adk.agents import SequentialAgent, LlmAgent
from google.adk.runners import Runner
from google.adk.sessions import InMemorySessionService
from google.genai import types

# Define tool functions for different stages
def extract_text(document_path: str) -> str:
    """Extracts text from a document file."""
    return f"This is sample content for processing from {document_path}."

def analyze_sentiment(text: str) -> dict:
    """Analyzes sentiment of the provided text."""
    positive = ["good", "great", "excellent"]
    negative = ["bad", "poor", "terrible"]
    lc = text.lower()
    pos = sum(w in lc for w in positive)
    neg = sum(w in lc for w in negative)
    tone = "positive" if pos > neg else "negative" if neg > pos else "neutral"
    return {"sentiment": tone, "positive_score": pos, "negative_score": neg}

def generate_summary(text: str, sentiment_data: dict) -> str:
    """Generates a summary based on text and sentiment data."""
    word_count = len(text.split())
    tone = sentiment_data["sentiment"]
    return f"Summary: {word_count} words, overall tone is {tone}."

# Create agents for each step
extract_agent = LlmAgent(
    name="extractor",
    model="gemini-2.0-flash",
    tools=[extract_text],
    output_key="extracted_text",
    description="Extracts text from the given document path.",
    instruction=(
        "You are the extractor. The user says 'Process this document: <path>'. "
        "Extract the <path> part, call `extract_text(path)`, and return only the extracted text."
    )
)
analyze_agent = LlmAgent(
    name="analyzer",
    model="gemini-2.0-flash",
    tools=[analyze_sentiment],
    description="Analyzes sentiment in text",
    output_key="sentiment_data",
    instruction=(
        "You are the analyzer. Take the text in `extracted_text`, "
        "call `analyze_sentiment(extracted_text)`, and return only the resulting JSON."
    )
)
summary_agent = LlmAgent(
    name="summarizer",
    model="gemini-2.0-flash",
    tools=[generate_summary],
    output_key="final_summary",
    description="Summarizes text with sentiment data.",
    instruction=(
        "You are the summarizer. Given `extracted_text` and `sentiment_data`, "
        "call `generate_summary(extracted_text, sentiment_data)` and return only the summary string."
    )
)
# Create the sequential pipeline
document_pipeline = SequentialAgent(
    name="document_processor",
    sub_agents=[extract_agent, analyze_agent, summary_agent],
    description="Extracts text, runs sentiment analysis, then summarizes."
)
# Set up runner and session
APP_NAME = "document_app"
USER_ID = "user_123"
SESSION_ID = "session_456"
# Create session service and session
session_service = InMemorySessionService()
session = session_service.create_session(
    app_name=APP_NAME,
    user_id=USER_ID,
    session_id=SESSION_ID
)
# Create runner
runner = Runner(
    agent=document_pipeline,
    app_name=APP_NAME,
    session_service=session_service
)

def run_document_pipeline(document_path: str) -> str:
    prompt = types.Content(
        role="user",
        parts=[types.Part(text=f"Process this document: {document_path}")]
    )
    final_summary = None
    for event in runner.run(user_id=USER_ID, session_id=SESSION_ID, new_message=prompt):
        # capture the summarizer's final output
        if event.is_final_response() and event.content and event.content.parts:
            final_summary = event.content.parts[0].text
        # keep looping until the generator naturally ends
    return (final_summary or "No response received.").strip()

# Example usage
result = run_document_pipeline("quarterly_report.pdf")
print(result)

2. Parallel Composition

When tasks are independent of each other, you can run them concurrently using a ParallelAgent to improve performance.

Example: Multi-Source Research Agent

from google.adk.agents import ParallelAgent, LlmAgent, SequentialAgent
from google.adk.runners import Runner
from google.adk.sessions import InMemorySessionService
from google.genai import types

# Define research tools for different sources
def search_news(topic: str) -> str:
    """Searches news sources for information on a topic."""
    return f"News results for {topic}: Latest developments include XYZ..."

def search_academic(topic: str) -> str:
    """Searches academic databases for information on a topic."""
    return f"Academic results for {topic}: Recent papers discuss ABC..."

def search_social(topic: str) -> str:
    """Searches social media for information on a topic."""
    return f"Social media trends for {topic}: People are discussing DEF..."

# Create agents for each research source
news_agent = LlmAgent(
    name="news_researcher",
    model="gemini-2.0-flash",
    tools=[search_news],
    output_key="news_results",
    description="Researches news sources",
    instruction=(
        "You are the news researcher. Extract the topic from the user query, "
        "call `search_news(topic)`, and return only the raw string."
    )
)
academic_agent = LlmAgent(
    name="academic_researcher",
    model="gemini-2.0-flash",
    tools=[search_academic],
    output_key="academic_results",
    description="Researches academic sources",
    instruction=(
        "You are the academic researcher. Extract the topic, "
        "call `search_academic(topic)`, and return only the raw string."
    )
)
social_agent = LlmAgent(
    name="social_researcher",
    model="gemini-2.0-flash",
    tools=[search_social],
    output_key="social_results",
    description="Researches social media trends",
    instruction=(
        "You are the social media researcher. Extract the topic, "
        "call `search_social(topic)`, and return only the raw string."
    )
)
# Create parallel research agent
parallel_research = ParallelAgent(
    name="parallel_researcher",
    sub_agents=[news_agent, academic_agent, social_agent],
    description="Fetches news, academic, and social results in parallel"
)
# Create a merger agent to combine results
def merge_research(news: str, academic: str, social: str) -> str:
    """Merges research results from multiple sources."""
    return f"""Combined Research Report:
News Insights: {news}
Academic Findings: {academic}
Social Trends: {social}
This comprehensive view provides a well-rounded perspective on the topic.
"""
merger_agent = LlmAgent(
    name="research_merger",
    model="gemini-2.0-flash",
    tools=[merge_research],
    output_key="merged_report",
    description="Merges research from multiple sources",
    instruction=(
        "You have three pieces of state: `news_results`, `academic_results`, "
        "and `social_results`. Call `merge_research(news_results, academic_results, social_results)` "
        "and return *only* the resulting combined report string."
    )
)
# Create a sequential pipeline that does parallel research then merges
research_pipeline = SequentialAgent(
    name="research_pipeline",
    sub_agents=[parallel_research, merger_agent],
    description="Runs parallel research and then merges the outputs"
)
# Set up runner and session
APP_NAME = "research_app"
USER_ID = "user_123"
SESSION_ID = "session_456"
# Create session service and session
session_service = InMemorySessionService()
session = session_service.create_session(
    app_name=APP_NAME,
    user_id=USER_ID,
    session_id=SESSION_ID
)
# Create runner
runner = Runner(
    agent=research_pipeline,
    app_name=APP_NAME,
    session_service=session_service
)

# Run the research pipeline
def run_research(topic: str) -> str:
    prompt = types.Content(
        role="user",
        parts=[types.Part(text=f"Research this topic: {topic}")]
    )
    merged_report = None
    # Fully consume the event stream
    for event in runner.run(user_id=USER_ID, session_id=SESSION_ID, new_message=prompt):
        # Only capture the merger agent's final response
        if (
            event.is_final_response()
            and event.author == "research_merger"
            and event.content and event.content.parts
        ):
            merged_report = event.content.parts[0].text
    return (merged_report or "No response received.").strip()

# Example usage
result = run_research("quantum computing")
print(result)

3. Iterative Refinement with LoopAgent

Use a LoopAgent when you need to repeatedly process or refine something until a condition is met.

Example: Iterative Content Refinement

from google.adk.agents import LoopAgent, LlmAgent, BaseAgent
from google.adk.runners import Runner
from google.adk.sessions import InMemorySessionService
from google.genai import types
from google.adk.events import Event, EventActions

# Define tools for content creation and critique
def generate_draft(topic: str) -> str:
    """Creates an initial draft on a topic."""
    return f"Initial draft about {topic}: This is a basic overview of the subject matter..."

def improve_draft(draft: str, feedback: str) -> str:
    """Improves a draft based on feedback."""
    # In a real implementation, this would make specific improvements
    return f"Improved draft based on feedback: {draft}\n\nAddressed issues: {feedback}"
# Create agents
writer_agent = LlmAgent(
    name="writer",
    model="gemini-2.0-flash",
    tools=[generate_draft, improve_draft],
    output_key="current_draft",
    description="Writes and refines content drafts",
    instruction=(
        "You are the writer. On the first iteration, extract the topic from the user message "
        "and call `generate_draft(topic)`. On subsequent iterations, take `current_draft` from state "
        "and `feedback` from state, call `improve_draft(current_draft, feedback)`. "
        "Return *only* the new draft string."
    )
)
class CriticAgent(BaseAgent):
    def __init__(self, name: str):
        super().__init__(name=name)
    async def _run_async_impl(self, context):
        state = context.session.state
        draft = state.get("current_draft", "")
        iteration = state.get("iteration", 0)
        # simple "quality" model: +25 points per revision
        quality = min(iteration * 25, 100)
        if quality >= 90:
            feedback = "Draft is excellent-no further changes needed."
            yield Event(
                author=self.name,
                content=types.Content(
                    role="assistant",
                    parts=[types.Part(text=feedback)]
                ),
                actions=EventActions(escalate=True)  # break the loop
            )
        else:
            feedback = f"Quality {quality}/100. Needs more clarity and detail."
            # bump iteration and store feedback
            yield Event(
                author=self.name,
                content=types.Content(
                    role="assistant",
                    parts=[types.Part(text=feedback)]
                ),
                actions=EventActions(state_delta={
                    "feedback": feedback,
                    "iteration": iteration + 1
                })
            )
critic_agent = CriticAgent(name="critic")
# Create loop agent for iterative refinement
content_refiner = LoopAgent(
    name="content_refiner",
    sub_agents=[writer_agent, critic_agent],
    max_iterations=5,
    description="Iteratively refines content until quality threshold is met"
)
# Set up runner and session
APP_NAME = "content_app"
USER_ID = "user_123"
SESSION_ID = "session_456"
# Create session service and session
session_service = InMemorySessionService()
session = session_service.create_session(
    app_name=APP_NAME,
    user_id=USER_ID,
    session_id=SESSION_ID
)
# Create runner
runner = Runner(
    agent=content_refiner,
    app_name=APP_NAME,
    session_service=session_service
)

# Run the content refiner
def run_content_refiner(topic: str) -> str:
    # seed the loop
    svc_session = session_service.get_session(app_name=APP_NAME, user_id=USER_ID, session_id=SESSION_ID)
    svc_session.state["iteration"] = 0
    prompt = types.Content(
        role="user",
        parts=[types.Part(text=f"Create high-quality content about: {topic}")]
    )
    final = None
    for event in runner.run(user_id=USER_ID, session_id=SESSION_ID, new_message=prompt):
        # capture only when the writer_agent emits a draft _and_ loop is about to end
        if event.is_final_response() and event.author == "writer":
            final = event.content.parts[0].text
    return (final or "No response received.").strip()

# Example usage
result = run_content_refiner("artificial intelligence ethics")
print(result)

In ADK, orchestration works by combining different types of agents and using the Runner system to manage execution flow.

Evaluation and Debugging in ADK

Once your agents are up and running, the next challenge is ensuring that they’re actually doing what you want them to do — reliably, safely, and efficiently. That’s where evaluation and debugging come in.

The Agent Development Kit (ADK) provides built-in mechanisms for tracing agent behavior, analyzing results, and building test cases. Combined with thoughtful observability practices, this helps you catch bugs, reduce hallucinations, and continuously refine your workflows.

Execution Traces

Every time an agent runs, ADK can log a trace — a step-by-step record of what happened. This includes:

• The agent’s input
• Which tool was selected (and why, if it was an LLM-based decision)
• Input/output to each tool
• Intermediate reasoning steps (e.g. LLM scratchpad)
• Final result

The Runner implementation handles this tracing automatically, and you can see the detailed events either through logging or by examining the returned events.

Debugging with Event Introspection

ADK provides tools for examining the event stream in detail, helping you debug complex agent behavior:

from google.adk.agents import LlmAgent
from google.adk.runners import Runner
from google.adk.sessions import InMemorySessionService
from google.genai import types
import json

# Create a simple agent that may have errors
def calculate_complex(expression: str) -> str:
    """Performs a complex calculation that might fail."""
    try:
        # Deliberately overcomplicated to demonstrate error handling
        parts = expression.split()
        result = eval(" ".join(parts))
        return str(result)
    except Exception as e:
        return f"Error: {str(e)}"

debug_agent = LlmAgent(
    name="debuggable_calculator",
    model="gemini-2.0-flash",
    tools=[calculate_complex],
    description="Performs calculations with detailed error handling for debugging"
)
# Set up runner and session
APP_NAME = "debug_app"
USER_ID = "debug_user"
SESSION_ID = "debug_session"
# Create session service and runner
session_service = InMemorySessionService()
session_service.create_session(
    app_name=APP_NAME,
    user_id=USER_ID,
    session_id=SESSION_ID
)
runner = Runner(
    agent=debug_agent,
    app_name=APP_NAME,
    session_service=session_service
)

# Run the agent with verbose debugging
def run_with_debugging(query):
    content = types.Content(
        role="user",
        parts=[types.Part(text=query)]
    )
    print("===== STARTING DEBUG SESSION =====")
    print(f"Input: {query}")
    print("===== EVENT STREAM =====")
    # Process every event in detail
    for i, event in enumerate(runner.run(
            user_id=USER_ID,
            session_id=SESSION_ID,
            new_message=content
    )):
        print(f"\nEVENT {i + 1}:")
        print(f"  Author: {event.author}")
        print(f"  ID: {event.id}")
        # Check for content
        if event.content:
            print(f"  Content Role: {event.content.role}")
            for j, part in enumerate(event.content.parts):
                print(f"  Content Part {j + 1}: {type(part).__name__}")
                if hasattr(part, 'text') and part.text:
                    print(f"    Text: {part.text[:100]}..." if len(part.text) > 100 else f"    Text: {part.text}")
        # Check for function calls
        function_calls = event.get_function_calls()
        if function_calls:
            print(f"  Function Calls: {len(function_calls)}")
            for fc in function_calls:
                print(f"    Name: {fc.name}")
                print(f"    Args: {json.dumps(fc.args, indent=2)}")
        # Check for function responses
        function_responses = event.get_function_responses()
        if function_responses:
            print(f"  Function Responses: {len(function_responses)}")
            for fr in function_responses:
                print(f"    Name: {fr.name}")
                print(f"    Response: {fr.response}")
        # Check for state changes
        if event.actions and event.actions.state_delta:
            print(f"  State Delta: {event.actions.state_delta}")
        # Check for final response
        if event.is_final_response():
            print("  [THIS IS FINAL RESPONSE]")
    print("===== DEBUG SESSION COMPLETE =====")

# Example usage
run_with_debugging("Calculate 10 * (5 + 3)")
run_with_debugging("Calculate 10 / (5 - 5)")  # Should trigger an error

Structured Event Stream

As we’ve seen in the examples, all agent interactions are managed through an event stream. The Runner returns these events, and you can use them to:

• Monitor agent progress in real-time
• Extract intermediate state changes
• Identify when the agent has completed its task

By processing this event stream, you can build sophisticated monitoring and debugging tools around your agents

Deployment Strategies: Running Agents in the Real World

Building intelligent agents in ADK is only the first step. For them to create real value, they need to be deployed — embedded into applications, run on a schedule, or served via APIs to users or other systems.

ADK offers a flexible deployment model that works equally well for:

• Local experimentation
• Cloud-native microservices
• Long-running workflows
• Serverless automation

Let’s explore the most common strategies.

1. Local Execution (Development & Prototyping)

Best for:

• Quick iteration
• Debugging
• Testing against live data

Agents can be run locally using the Runner pattern we’ve seen throughout this guide:

from google.adk.agents import LlmAgent
from google.adk.runners import Runner
from google.adk.sessions import InMemorySessionService
from google.genai import types

# Create a simple agent
def answer_question(question: str) -> str:
    """Provides an answer to a question."""
    return f"The answer to '{question}' is: 42"
agent = LlmAgent(
    name="simple_agent",
    model="gemini-2.0-flash",
    tools=[answer_question],
    description="A simple question-answering agent"
)
# Set up runner with InMemorySessionService for local use
session_service = InMemorySessionService()
runner = Runner(
    agent=agent,
    app_name="local_app",
    session_service=session_service
)
# Function to run the agent locally
def run_local_query(query):
    # Create session for this run
    session = session_service.create_session(
        app_name="local_app",
        user_id="local_user",
        session_id="local_session"
    )
    # Create content from query
    content = types.Content(
        role="user",
        parts=[types.Part(text=query)]
    )
    # Run the agent and extract the final response
    for event in runner.run(
        user_id="local_user",
        session_id="local_session",
        new_message=content
    ):
        if event.is_final_response():
            return event.content.parts[0].text
    return "No response received."
# Example usage in a local script
if __name__ == "__main__":
    while True:
        user_input = input("Ask a question (or 'exit' to quit): ")
        if user_input.lower() == 'exit':
            break
        response = run_local_query(user_input)
        print(f"Agent: {response}\n")

2. REST API with FastAPI or Flask

Best for:

• Deploying agents as services
• Integrating with web apps or backends

Wrap your agent in a web server and expose it as an HTTP endpoint:

from fastapi import FastAPI, WebSocket, WebSocketDisconnect
from fastapi.responses import JSONResponse
from pydantic import BaseModel
import uuid
import uvicorn

from google.adk.agents import LlmAgent
from google.adk.runners import Runner
from google.adk.sessions import InMemorySessionService
from google.genai import types
# --- Agent setup ----------------------------------------------------------
def answer_question(question: str) -> str:
    """Provides an answer to a question."""
    return f"The answer to '{question}' is: 42"
agent = LlmAgent(
    name="api_agent",
    model="gemini-2.0-flash",
    tools=[answer_question],
    description="A question-answering agent exposed as an API"
)
session_service = InMemorySessionService()
runner = Runner(
    agent=agent,
    app_name="api_app",
    session_service=session_service
)
# --- Request/Response models ----------------------------------------------
class QueryRequest(BaseModel):
    query: str
    session_id: str = None
    user_id: str = None
class QueryResponse(BaseModel):
    user_id: str
    session_id: str
    response: str
# --- FastAPI app ----------------------------------------------------------
app = FastAPI(title="ADK Agent API")
@app.post("/query", response_model=QueryResponse)
async def query_agent(req: QueryRequest):
    # generate IDs if missing
    user_id = req.user_id or f"user_{uuid.uuid4().hex[:8]}"
    session_id = req.session_id or f"session_{uuid.uuid4().hex[:8]}"
    # ensure session exists
    if session_service.get_session(app_name="api_app", user_id=user_id, session_id= session_id) is None:
        session_service.create_session(app_name="api_app", user_id=user_id, session_id= session_id)
    # build message
    content = types.Content(
        role="user",
        parts=[types.Part(text=req.query)]
    )
    # stream until final and grab the text
    response_text = None
    async for event in runner.run_async(
        user_id=user_id,
        session_id=session_id,
        new_message=content
    ):
        if event.is_final_response() and event.content and event.content.parts:
            response_text = event.content.parts[0].text
    return QueryResponse(
        user_id=user_id,
        session_id=session_id,
        response=response_text or ""
    )
@app.websocket("/ws/{session_id}")
async def websocket_endpoint(ws: WebSocket, session_id: str):
    await ws.accept()
    # each WS connection gets its own user_id
    user_id = f"ws_user_{uuid.uuid4().hex[:8]}"
    # ensure session exists
    if session_service.get_session(app_name="api_app", user_id=user_id, session_id= session_id is None):
        session_service.create_session(app_name="api_app", user_id=user_id, session_id= session_id)
    # let client know its user_id
    await ws.send_json({"type": "session_init", "user_id": user_id, "session_id": session_id})
    try:
        while True:
            query = await ws.receive_text()
            content = types.Content(
                role="user",
                parts=[types.Part(text=query)]
            )
            # stream all events, pushing each as JSON
            async for event in runner.run_async(
                user_id=user_id,
                session_id=session_id,
                new_message=content
            ):
                payload = {
                    "type": "event",
                    "id": event.id,
                    "author": event.author,
                }
                if event.content and event.content.parts:
                    payload["text"] = event.content.parts[0].text
                if event.is_final_response():
                    payload["is_final"] = True
                await ws.send_json(payload)
    except WebSocketDisconnect:
        # client closed connection
        pass
    except Exception as e:
        # on error, close cleanly
        await ws.send_json({"type": "error", "message": str(e)})
        await ws.close()
if __name__ == "__main__":
    uvicorn.run(app, host="0.0.0.0", port=8000)

3. Google Vertex AI Agent Engine

For production deployment within Google Cloud, you can use the Vertex AI Agent Engine, which is specifically designed to work with ADK. This provides:

• Fully managed infrastructure
• Scalable endpoints
• Native integration with Google Cloud services

from google.adk.agents import LlmAgent
from vertexai.preview.reasoning_engines import AdkApp

# Create an agent
def answer_question(question: str) -> str:
    """Provides an answer to a question."""
    return f"The answer to '{question}' is: 42"
agent = LlmAgent(
    name="vertex_agent",
    model="gemini-2.0-flash",
    tools=[answer_question],
    description="A question-answering agent deployed on Vertex AI"
)
# Create Vertex AI app from the agent
app = AdkApp(agent=agent)
# For local testing with Vertex AI integration
for event in app.stream_query(
    user_id="test_user",
    message="What is the meaning of life?"
):
    print(event)
# For deployment to Vertex AI Agent Engine
# app.deploy(project="your-gcp-project", display_name="Question Answerer")

4. Scheduled or Event-Driven Execution

You can run agents on a schedule or trigger them based on external events:

from google.adk.agents import LlmAgent
from google.adk.runners import Runner
from google.adk.sessions import InMemorySessionService
from google.genai import types
import time
import schedule
import json

# Create an agent that summarizes data
def summarize_data(source: str) -> str:
    """Summarizes data from a source."""
    # In a real implementation, this would fetch and analyze data
    return f"Summary of data from {source}: Key metrics are trending upward."
summary_agent = LlmAgent(
    name="summary_agent",
    model="gemini-2.0-flash",
    tools=[summarize_data],
    description="An agent that summarizes data on a schedule"
)
# Set up runner
session_service = InMemorySessionService()
runner = Runner(
    agent=summary_agent,
    app_name="schedule_app",
    session_service=session_service
)
# Function to run the summary job
def run_summary_job():
    print(f"Running scheduled summary job at {time.strftime('%Y-%m-%d %H:%M:%S')}")
    # Create a unique session for this run
    session_id = f"job_{int(time.time())}"
    session = session_service.create_session(
        app_name="schedule_app",
        user_id="schedule_user",
        session_id=session_id
    )
    # Create content with the job request
    content = types.Content(
        role="user",
        parts=[types.Part(text="Summarize today's data from our analytics database")]
    )
    # Run the agent
    summary = None
    for event in runner.run(
        user_id="schedule_user",
        session_id=session_id,
        new_message=content
    ):
        if event.is_final_response():
            summary = event.content.parts[0].text
    if summary:
        # In a real implementation, you might:
        # - Send an email with the summary
        # - Store it in a database
        # - Trigger an alert if certain conditions are met
        print(f"Summary generated: {summary}")
        # Example: Write to a log file
        with open("summaries.log", "a") as log_file:
            log_entry = {
                "timestamp": time.strftime("%Y-%m-%d %H:%M:%S"),
                "session_id": session_id,
                "summary": summary
            }
            log_file.write(json.dumps(log_entry) + "\n")
    else:
        print("No summary generated.")
# Schedule the job to run daily at 8:00 AM
schedule.every().day.at("08:00").do(run_summary_job)
# Run the scheduler
if __name__ == "__main__":
    print("Starting scheduler...")
    # Run once immediately for testing
    run_summary_job()
    # Then run on schedule
    while True:
        schedule.run_pending()
        time.sleep(60)  # Check every minute

5. Containerized Deployment (Docker + Orchestration)

For full control and portability, containerize your agent service:

FROM python:3.10-slim

# 1. Create a non-root user
RUN addgroup --system appgroup && \
    adduser --system appuser --ingroup appgroup
WORKDIR /app
# 2. Install dependencies
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
# 3. Copy code
COPY . .
# 4. Drop to non-root
USER appuser
# 5. Environment vars (optional defaults)
ENV APP_NAME="docker_app" \
    MODEL_NAME="gemini-2.0-flash"
# 6. Expose the FastAPI port
EXPOSE 8080
# 7. Launch via Uvicorn
CMD ["uvicorn", "api_server:app", "--host", "0.0.0.0", "--port", "8080"]

With the corresponding api_server.py file containing your ADK agent setup and API endpoints. You can then deploy this container to:

• Kubernetes
• Google Cloud Run
• AWS ECS
• Azure Container Apps

Responsible Agent Design: Building Agents You Can Trust

As language model–powered agents become more capable and autonomous, the importance of responsible design can’t be overstated. It’s no longer just about what your agent can do — it’s about what it should do, and how you ensure it behaves reliably, safely, and ethically.

Google’s ADK bakes this philosophy into its structure. It doesn’t enforce rules, but it gives you the tools to build agents that are transparent, controllable, auditable, and aligned with real-world values.

This section walks through practical design choices you can make to ensure your agents don’t just work — they deserve to be trusted.

1. Be Explicit About Capabilities and Limits

Agents that use LLMs can come off as all-knowing — but they aren’t. They generate confident-sounding answers even when they’re guessing. That’s dangerous if left unchecked.

Best practices:

• Use prompts or preambles that remind the model of its limitations.
• Include disclaimers in outputs for uncertain or generative tasks.
• Set clear expectations: is this a search assistant, or a legal advisor? (Hopefully not the latter.)

# Example of setting clear limitations in the agent instruction
medical_info_agent = LlmAgent(
    name="health_info",
    model="gemini-2.0-flash",
    instruction="""You provide general health information only. 
    You are NOT a doctor and cannot diagnose conditions or recommend treatments.
    Always clarify that you're providing general information, not medical advice.
    If asked for medical advice, recommend consulting a healthcare professional.
    """,
    description="Provides general health information with appropriate disclaimers"
)

2. Design for Tool Safety

Tools are powerful — they can make API calls, manipulate data, or even trigger other agents. Make sure they can’t be misused.

Tips:

• Validate inputs before running tools.
• Add input/output schemas where possible.
• Log every tool invocation for traceability.

# Example of a tool with built-in safety checks
def send_email(to: str, subject: str, body: str) -> dict:
    """Sends an email to the specified recipient.
        Args:
        to: Email address of the recipient.
        subject: Subject line of the email.
        body: Body text of the email.
    Returns:
        A dictionary with status information.
    """
    # Safety check: Validate email format
    if not re.match(r"[^@]+@[^@]+\.[^@]+", to):
        return {"status": "error", "message": "Invalid email format"}
    # Safety check: Check allowed domains
    allowed_domains = ["mycompany.com", "partner.org"]
    if not any(to.endswith(f"@{domain}") for domain in allowed_domains):
        return {
            "status": "error", 
            "message": f"Can only send to these domains: {', '.join(allowed_domains)}"
        }
    # Safety check: Check for sensitive content
    sensitive_terms = ["password", "ssn", "secret", "confidential"]
    for term in sensitive_terms:
        if term in body.lower():
            return {
                "status": "error", 
                "message": f"Cannot send emails containing sensitive terms: {term}"
            }
    # Actual email sending logic would go here
    # ...
    # Log the action for audit purposes
    logging.info(f"Email sent to {to} with subject: {subject}")
    return {"status": "success", "message": "Email sent successfully"}

3. Add Human-in-the-Loop Checks

Not all decisions should be fully automated. Some need a human review step, especially if:

• The agent interacts with customers
• The stakes are high (legal, medical, financial)
• You’re still evaluating its reliability

Example of a human approval tool:

from google.adk.agents import LlmAgent
from google.adk.runners import Runner
from google.adk.sessions import InMemorySessionService
from google.genai import types

# Tool that requires human approval
def propose_action(action_type: str, details: str) -> dict:
    """Proposes an action that requires human approval.
    Args:
        action_type: The type of action being proposed.
        details: Specific details about the action.
    Returns:
        A dictionary with the proposal status.
    """
    # In a real implementation, this might:
    # - Send the proposal to a queue for human review
    # - Send an email notification
    # - Update a dashboard
    print(f"\n=== ACTION REQUIRING APPROVAL ===")
    print(f"Type: {action_type}")
    print(f"Details: {details}")
    # Simple console-based approval for demo purposes
    approval = input("Approve? (y/n): ")
    if approval.lower() == 'y':
        return {
            "status": "approved",
            "message": "Action approved by human reviewer",
            "action_type": action_type,
            "details": details
        }
    else:
        return {
            "status": "rejected",
            "message": "Action rejected by human reviewer",
            "action_type": action_type,
            "details": details
        }
# Create agent with human-in-the-loop design
human_oversight_agent = LlmAgent(
    name="overseen_agent",
    model="gemini-2.0-flash",
    tools=[propose_action],
    instruction="""You are an agent with human oversight.
    For any significant action, use the propose_action tool to get approval before proceeding.
    Significant actions include:
    - Sending communications to customers
    - Making financial decisions above $100
    - Changing system settings
    - Creating or deleting accounts
    """,
    description="Agent that requires human approval for significant actions"
)
# Set up runner
session_service = InMemorySessionService()
runner = Runner(
    agent=human_oversight_agent,
    app_name="oversight_app",
    session_service=session_service
)
# Function to run the agent
def run_overseen_agent(query):
    # Create a session
    session = session_service.create_session(
        app_name="oversight_app",
        user_id="user_123",
        session_id="session_456"
    )
    # Create content
    content = types.Content(
        role="user",
        parts=[types.Part(text=query)]
    )
    # Run the agent
    for event in runner.run(
        user_id="user_123",
        session_id="session_456",
        new_message=content
    ):
        if event.is_final_response():
            return event.content.parts[0].text
    return "No response received."
# Example usage
result = run_overseen_agent("Please send a promotional email to all customers about our new product.")
print(f"\nFinal result: {result}")

4. Monitor and Improve

Responsibility isn’t a one-time setup — it’s ongoing. You should:

• Track how your agent performs over time
• Gather user feedback (especially about wrong or weird outputs)
• Tune prompts or tool logic as things evolve

ADK’s evaluation framework is a great way to build a feedback loop into your development process.

By thinking through these dimensions from the start, you’ll not only build agents that work — you’ll build agents that people actually trust, and that you feel confident deploying in the real world.

Conclusion

As the field of agentic AI evolves, tools like Google’s Agent Development Kit (ADK) offer a compelling blueprint for how we can move from isolated, one-off prompts to structured, intelligent systems. ADK is not just a wrapper around language models — it’s a full ecosystem for composing agents, orchestrating workflows, evaluating behaviors, and deploying trustworthy applications into the real world.

The key takeaways from our exploration are:

1. ADK provides structured agent types (LlmAgent, Sequential, Parallel, Loop) for different use cases
2. Tools are created by simply defining Python functions with good documentation
3. The Runner pattern is essential for executing agents properly
4. Orchestration capabilities let you build sophisticated multi-agent systems
5. Event-based architecture enables detailed monitoring and debugging

Whether you’re building a personal AI assistant, a business process automation system, or an experimental research agent, ADK gives you the right primitives to move fast — without sacrificing structure or control.

What makes ADK especially promising is how it blends developer ergonomics with agent design principles. You can start small with a few Python functions and build up to complex, multi-agent systems — all while keeping a clear mental model of how things work under the hood.

Happy building! 🎉