google-adk-redis 0.1.5

Redis-backed session service for Google ADK

google-adk-redis

Redis-backed session service for Google ADK.

Install

From a Git repo:

pip install git+https://github.com/hucruz/google-adk-redis.git

From PyPI:

pip install google-adk-redis

Usage (programmatic)

from google_adk_redis import RedisMemorySessionService

session_service = RedisMemorySessionService(
    host="localhost",
    port=6379,
    db=0,
    expire=60 * 60,
    max_transaction_retries=16,
)

Instances created with the same Redis configuration reuse the same underlying async Redis client by default, so creating many services does not multiply connection pools. You can opt out with share_cache=False or inject a custom client with cache=....

If your application may attempt to load and create the same known session ID concurrently, prefer get_or_create_session(...) over a manual get_session(...) then create_session(...) sequence.

Each session is stored in its own Redis key using the shape session:<app_name>:<user_id>:<session_id>, so different sessions for the same user no longer overwrite each other through a shared JSON blob. If you still expect concurrent writes to the same session across processes, increase max_transaction_retries and, if needed, adjust transaction_retry_base_delay / transaction_retry_max_delay.

Optional register for google.adk.sessions

import google_adk_redis

google_adk_redis.register()
from google.adk.sessions import RedisMemorySessionService

Usage with Google ADK (Runner)

import asyncio

from google.adk.agents.llm_agent import LlmAgent
from google.adk.runners import Runner
from google.genai import types

from google_adk_redis import RedisMemorySessionService


async def main():
    session_service = RedisMemorySessionService(
        host="localhost",
        port=6379,
        db=0,
        expire=60 * 60,
    )
    agent = LlmAgent(
        name="assistant",
        model="gemini-2.0-flash",
        instruction="You are a helpful assistant.",
    )

    async with Runner(
        app_name="demo-app",
        agent=agent,
        session_service=session_service,
    ) as runner:
        session = await session_service.create_session(
            app_name="demo-app",
            user_id="user-123",
        )

        async for event in runner.run_async(
            user_id=session.user_id,
            session_id=session.id,
            new_message=types.Content(
                role="user",
                parts=[types.Part(text="Hola!")],
            ),
        ):
            if event.content and event.content.parts:
                text = "".join(part.text or "" for part in event.content.parts)
                if text:
                    print(text)


asyncio.run(main())

Authors

• BloodBoy21 (original implementation, https://github.com/BloodBoy21)
• hucruz (upstream contributions, https://github.com/hucruz)

Maintainer

• hucruz (https://github.com/hucruz)

Attribution

Source repository: https://github.com/BloodBoy21/nerds-adk-python/tree/feat-redis-session

This repository packages and adapts the upstream implementation for standalone use.