Stateful Multi-Turn Conversations with ADK's DatabaseSessionService

One of the most challenging obstacles in building production AI agents is state persistence. Without persistent state management, every incoming HTTP request starts from a blank slate—leaving your agent unable to recall past user choices, active shopping cart contents, or intermediate transaction details across turns.

Google ADK solves this challenge through its DatabaseSessionService component, which serializes conversation turns and state variables into SQL databases. In this guide, we will walk through building a stateful shopping cart assistant that maintains state reliably, even across server restarts or pod re-deployments.

---

Why In-Memory State Is Not Enough for Production

During initial prototyping, ADK’s InMemorySessionService is convenient because it stores state directly in Python RAM. However, in containerized cloud environments (such as Cloud Run or Kubernetes), user traffic is load-balanced across multiple instances:

• Turn #1 (User: “Add 2 books to my cart”) hits Container Instance A.
• Turn #2 (User: “What is in my cart?”) hits Container Instance B.

If state is stored in memory, Instance B has no record of Turn #1. DatabaseSessionService eliminates this failure mode by decoupling session memory from container lifecycles and centralizing state in a durable SQL backend.

---

Building a Stateful Shopping Cart Agent

1. Define a State-Aware Function Tool

To read or update state within a tool function, declare a callback_context: CallbackContext parameter. Google ADK automatically injects the active session context into this parameter during tool invocation.

import asyncio
from google.adk.agents import Agent
from google.adk.tools import FunctionTool
from google.adk.agents.callback_context import CallbackContext
from google.adk.runners import Runner
from google.adk.sessions import DatabaseSessionService

def add_item_to_cart(item_name: str, quantity: int, callback_context: CallbackContext) -> dict:
    """Adds a requested item and quantity to the user's persistent shopping cart.

    Args:
        item_name: The name of the product (e.g. 'book', 'pen').
        quantity: The quantity of the product to add.
    """
    # Access persistent session state dictionary
    state = callback_context.state
    if "cart" not in state:
        state["cart"] = {}

    current_qty = state["cart"].get(item_name, 0)
    state["cart"][item_name] = current_qty + quantity

    return {
        "status": "success",
        "updated_cart": state["cart"]
    }

cart_tool = FunctionTool(func=add_item_to_cart)

---

2. Configure Agent and DatabaseSessionService

Next, instantiate the cart_agent and connect it to a SQLite database file using DatabaseSessionService:

cart_agent = Agent(
    name="shopping_assistant",
    model="gemini-2.5-flash",
    instruction=(
        "You are a helpful shopping assistant. "
        "Help users manage their shopping cart using `add_item_to_cart`."
    ),
    tools=[cart_tool]
)

async def main():
    # Store sessions persistently in local SQLite database 'sessions.db'
    db_uri = "sqlite:///sessions.db"
    session_service = DatabaseSessionService(db_uri=db_uri)

    runner = Runner(agent=cart_agent, session_service=session_service)
    session_id = "user-cart-session-999"

    print("--- Turn 1 ---")
    res1 = await runner.run_async(
        session_id=session_id, 
        message="Please add 2 books to my cart."
    )
    print("Agent Response:\n", res1.text)

    print("\n--- Turn 2 ---")
    res2 = await runner.run_async(
        session_id=session_id, 
        message="Add 1 pen as well. What is in my cart now?"
    )
    print("Agent Response:\n", res2.text)

if __name__ == "__main__":
    asyncio.run(main())

---

Output Verification

When executed, the agent harness retains memory between multi-turn calls:

--- Turn 1 ---
Agent Response:
I've added 2 books to your cart.

--- Turn 2 ---
Agent Response:
I have added 1 pen to your cart. Your current cart contains:
- Books: 2
- Pens: 1

If you stop the Python script after Turn 1, restart your terminal, and run Turn 2, the agent will still accurately report the 2 books previously saved in sessions.db.

---

Production Database Configuration

For cloud production environments, transition from local SQLite URIs to enterprise relational database connections by updating the SQLAlchemy connection string:

• PostgreSQL: postgresql+psycopg2://user:password@pg-host:5432/agent_sessions
• Cloud Spanner: spanner://projects/my-project/instances/my-instance/databases/my-db
• MySQL / MariaDB: mysql+pymysql://user:password@mysql-host:3306/agent_sessions

---

The ADK Session Lifecycle

Understanding how ADK manages state during a turn cycle clarifies debugging and architecture choices:

flowchart TD
    A[1. Runner receives run_async with session_id] --> B[2. Session history & state dict loaded from Database]
    B --> C[3. Agent calls model & executes tool functions]
    C --> D[4. Tools & Callbacks mutate callback_context.state]
    D --> E[5. Runner persists updated turns & state to Database]
    E --> F[6. Return final response to client]

1. Retrieval: Upon receiving run_async(session_id=...), the Runner queries the database to reconstruct past messages and load state.
2. Execution: System prompts, conversation history, and tool definitions are transmitted to Gemini.
3. State Mutation: Callback hooks or function tools read/write callback_context.state.
4. Transaction Commit: At turn conclusion, ADK commits updated event logs and state variables back to the database.

---

Key Takeaways

Using DatabaseSessionService is a fundamental requirement when moving AI agents into production. By storing session data in durable databases rather than process memory, your Google ADK agent harness gains seamless horizontal scalability, resilience against process restarts, and multi-turn state consistency.