This note contains an example of how to manage sessions with Google Agent Developer Kit (ADK) in Python. It pulls heavily from this Kaggle notebook
Setup
First, we import libraries.
from typing import Any, Dict
from google.adk.agents import Agent, LlmAgent
from google.adk.apps.app import App, EventsCompactionConfig
from google.adk.models.google_llm import Gemini
from google.adk.sessions import DatabaseSessionService
from google.adk.sessions import InMemorySessionService
from google.adk.runners import Runner
from google.adk.tools.tool_context import ToolContext
from google.genai import typesThen we define a helper function to run a session:
# Define helper functions that will be reused throughout the notebook
async def run_session(
runner_instance: Runner,
user_queries: list[str] | str = None,
session_name: str = "default",
):
print(f"\n ### Session: {session_name}")
# Get app name from the Runner
app_name = runner_instance.app_name
# Attempt to create a new session or retrieve an existing one
try:
session = await session_service.create_session(
app_name=app_name, user_id=USER_ID, session_id=session_name
)
except:
session = await session_service.get_session(
app_name=app_name, user_id=USER_ID, session_id=session_name
)
# Process queries if provided
if user_queries:
# Convert single query to list for uniform processing
if type(user_queries) == str:
user_queries = [user_queries]
# Process each query in the list sequentially
for query in user_queries:
print(f"\nUser > {query}")
# Convert the query string to the ADK Content format
query = types.Content(role="user", parts=[types.Part(text=query)])
# Stream the agent's response asynchronously
async for event in runner_instance.run_async(
user_id=USER_ID, session_id=session.id, new_message=query
):
# Check if the event contains valid content
if event.content and event.content.parts:
# Filter out empty or "None" responses before printing
if (
event.content.parts[0].text != "None"
and event.content.parts[0].text
):
print(f"{MODEL_NAME} > ", event.content.parts[0].text)
else:
print("No queries!")And we configure retry options, since we’re hitting an external tool:
retry_config = types.HttpRetryOptions(
attempts=5, # Maximum retry attempts
exp_base=7, # Delay multiplier
initial_delay=1,
http_status_codes=[429, 500, 503, 504], # Retry on these HTTP errors
)Creating a Stateful Agent
Next we’ll create a stateful agent using ADK. We’ll just manage the session in memory for now, but there are other options:
APP_NAME = "default" # Application
USER_ID = "default" # User
SESSION = "default" # Session
MODEL_NAME = "gemini-2.5-flash-lite"
# Step 1: Create the LLM Agent
root_agent = Agent(
model=Gemini(model="gemini-2.5-flash-lite", retry_options=retry_config),
name="text_chat_bot",
description="A text chatbot", # Description of the agent's purpose
)
# Step 2: Set up Session Management
# InMemorySessionService stores conversations in RAM (temporary)
session_service = InMemorySessionService()
# Step 3: Create the Runner
runner = Runner(agent=root_agent, app_name=APP_NAME, session_service=session_service)We can then prompt it with a few things:
# Run a conversation with two queries in the same session
# Notice: Both queries are part of the SAME session, so context is maintained
await run_session(
runner,
[
"Hi, I am Sam! What is the capital of United States?",
"Hello! What is my name?", # This time, the agent should remember!
],
"stateful-agentic-session",
)And if we want to see if it can remember what we asked earlier, we can:
# Run this cell after restarting the kernel. All this history will be gone...
await run_session(
runner,
["What did I ask you about earlier?", "And remind me, what's my name?"],
"stateful-agentic-session",
) # Note, we are using same session nameCreating Persistent Sessions
Using in-memory sessions is fine for prototyping, but it’s typically not the best approach for production applications. ADK offers a handful of different SessionService implementations depending on a user’s needs:
| Service | Use Case | Persistence | Best For |
|---|---|---|---|
| InMemorySessionService | Development & Testing | ❌ Lost on restart | Quick prototypes |
| DatabaseSessionService | Self-managed apps | ✅ Survives restarts | Small to medium apps |
| Agent Engine Sessions | Production on GCP | ✅ Fully managed | Enterprise scale |
If we want to create a DatabaseSessionService using SQLite, we can set this up as follows: |
# Step 1: Create the same agent (notice we use LlmAgent this time)
chatbot_agent = LlmAgent(
model=Gemini(model="gemini-2.5-flash-lite", retry_options=retry_config),
name="text_chat_bot",
description="A text chatbot with persistent memory",
)
# Step 2: Switch to DatabaseSessionService
# SQLite database will be created automatically
db_url = "sqlite:///my_agent_data.db" # Local SQLite file
session_service = DatabaseSessionService(db_url=db_url)
# Step 3: Create a new runner with persistent storage
runner = Runner(agent=chatbot_agent, app_name=APP_NAME, session_service=session_service)And we can interact with the model again:
await run_session(
runner,
["Hi, I am Sam! What is the capital of the United States?", "Hello! What is my name?"],
"test-db-session-01",
)If we want to see how the data is stored in the database, we can do that via:
import sqlite3
def check_data_in_db():
with sqlite3.connect("my_agent_data.db") as connection:
cursor = connection.cursor()
result = cursor.execute(
"select app_name, session_id, author, content from events"
)
print([_[0] for _ in result.description])
for each in result.fetchall():
print(each)
check_data_in_db()Context Compaction
ADK also supports compacting context/events stored in the session. This summarizes past events, making the context easier for the agent to work with. Here’s how we would implement this:
# Re-define our app with Events Compaction enabled
research_app_compacting = App(
name="research_app_compacting",
root_agent=chatbot_agent,
# This is the new part!
events_compaction_config=EventsCompactionConfig(
compaction_interval=3, # Trigger compaction every 3 invocations
overlap_size=1, # Keep 1 previous turn for context
),
)
db_url = "sqlite:///my_agent_data.db" # Local SQLite file
session_service = DatabaseSessionService(db_url=db_url)
# Create a new runner for our upgraded app
research_runner_compacting = Runner(
app=research_app_compacting, session_service=session_service
)