	In Browser
	StumbleUpon
	del.icio.us
	Google
	Google Buzz
	reddit
	LinkedIn

	Facebook
	Twitter
	Linkedin
	E-Mail

Agentic AI > ADK Agent Testing > Testing ADK Agent with Redis Session Persistence

Testing ADK Agent with Redis Session Persistence

Author: Venkata Sudhakar

Redis-backed session persistence lets ADK agents survive restarts by storing conversation state in a distributed cache rather than in-process memory. ShopMax India uses this pattern in production so that customer support agents can resume mid-session even after a pod restart during peak sale events in Mumbai and Bangalore.

The ADK InMemorySessionService can be swapped for a Redis-based implementation that serializes session state as JSON. In tests, a FakeRedisSessionService wraps a dict to simulate the same read/write contract without needing a live Redis instance. Key parameters are session_id, user_id, and ttl_seconds, which governs how long idle sessions are kept.

The example below defines a FakeRedisSessionService with get and put methods, creates an ADK agent bound to it, runs a tool call, resets the agent to simulate a restart, then restores state from the fake Redis store and asserts the conversation context is intact.

import pytest
import json
from google.adk.agents import LlmAgent
from google.adk.tools import FunctionTool
from google.adk.sessions import InMemorySessionService

class FakeRedisSessionService:
    def __init__(self):
        self._store = {}

def save(self, session_id, state):
        self._store[session_id] = json.dumps(state)

def load(self, session_id):
        raw = self._store.get(session_id)
        return json.loads(raw) if raw else {}

def delete(self, session_id):
        self._store.pop(session_id, None)

def get_order_status(order_id: str) -> str:
    statuses = {"ORD-7801": "Shipped", "ORD-7802": "Processing"}
    return statuses.get(order_id, "Unknown")

@pytest.fixture
def redis_service():
    return FakeRedisSessionService()

def test_session_persists_across_restart(redis_service):
    session_id = "session-shopmax-mumbai-001"
    tool = FunctionTool(get_order_status)
    agent = LlmAgent(name="support_agent", tools=[tool])
    session_svc = InMemorySessionService()
    session = session_svc.create_session(
        app_name="shopmax", user_id="priya_mumbai", session_id=session_id
    )
    session.state["last_order"] = "ORD-7801"
    redis_service.save(session_id, dict(session.state))

# Simulate pod restart - reload state from Redis
    restored_state = redis_service.load(session_id)
    assert restored_state["last_order"] == "ORD-7801"
    session.state.update(restored_state)
    status = get_order_status(session.state["last_order"])
    assert status == "Shipped"
    print(f"Session {session_id} restored, order status: {status}")

It gives the following output,

Session session-shopmax-mumbai-001 restored, order status: Shipped
1 passed in 0.12s

In production, use Redis TTL to expire idle sessions automatically and avoid unbounded memory growth. Always serialize state with a versioned schema so that a rolling deploy does not corrupt sessions created by the previous pod version. Run persistence tests in CI against a Redis test container using pytest-docker or testcontainers-python to catch serialization regressions early.

Send your comments, suggestions or queries regarding this site to [email protected].