sessions

sessions

Redis session services for ADK.

RedisSessionMemoryService

RedisSessionMemoryService(config: RedisSessionMemoryServiceConfig | None = None)

Bases: BaseSessionService

Session service backed by Redis memory backends.

Redis Agent Memory stores session events by session ID. The self-hosted Agent Memory Server stores sessions in Working Memory. Returned ADK sessions keep the original caller-facing session ID for both backends.

Initialize the Redis Agent Memory session service.

Parameters:

Name	Type	Description	Default
config	RedisSessionMemoryServiceConfig | None	Configuration for the service. If None, defaults are used.	None

Source code in src/adk_redis/sessions/session_memory.py

def __init__(self, config: RedisSessionMemoryServiceConfig | None = None):
  """Initialize the Redis Agent Memory session service.

  Args:
      config: Configuration for the service. If None, defaults are used.
  """
  self._config = config or RedisSessionMemoryServiceConfig()

create_session async

create_session(*, app_name: str, user_id: str, state: dict[str, Any] | None = None, session_id: str | None = None) -> Session

Create a new ADK session object.

Redis Agent Memory creates a stored session when the first event is appended, so this method returns the ADK session without writing a marker event.

Source code in src/adk_redis/sessions/session_memory.py

@override
async def create_session(
    self,
    *,
    app_name: str,
    user_id: str,
    state: dict[str, Any] | None = None,
    session_id: str | None = None,
) -> Session:
  """Create a new ADK session object.

  Redis Agent Memory creates a stored session when the first event is
  appended, so this method returns the ADK session without writing a marker
  event.
  """
  if self._config.backend == OPENSOURCE_AGENT_MEMORY_BACKEND:
    return await self._create_session_agent_memory_server(
        app_name=app_name,
        user_id=user_id,
        state=state,
        session_id=session_id,
    )

  session_id = (
      session_id.strip()
      if session_id and session_id.strip()
      else str(uuid.uuid4())
  )
  return Session(
      id=session_id,
      app_name=app_name,
      user_id=user_id,
      state=state or {},
      events=[],
      last_update_time=time.time(),
  )

get_session async

get_session(*, app_name: str, user_id: str, session_id: str, config: GetSessionConfig | None = None) -> Session | None

Retrieve an ADK session from Redis Agent Memory.

Source code in src/adk_redis/sessions/session_memory.py

@override
async def get_session(
    self,
    *,
    app_name: str,
    user_id: str,
    session_id: str,
    config: GetSessionConfig | None = None,
) -> Session | None:
  """Retrieve an ADK session from Redis Agent Memory."""
  if self._config.backend == OPENSOURCE_AGENT_MEMORY_BACKEND:
    return await self._get_session_agent_memory_server(
        app_name=app_name,
        user_id=user_id,
        session_id=session_id,
        config=config,
    )

  storage_session_id = self._storage_session_id(
      app_name=app_name,
      user_id=user_id,
      session_id=session_id,
  )

  try:
    async with self._agent_memory() as agent_memory:
      response = await agent_memory.get_session_memory_async(
          session_id=storage_session_id
      )
  except Exception as e:
    if is_not_found_error(e):
      # Managed sessions are materialized on first append_event. Return an
      # empty ADK session so runners can start a new conversation.
      return Session(
          id=session_id,
          app_name=app_name,
          user_id=user_id,
          state={},
          events=[],
          last_update_time=time.time(),
      )
    logger.error("Failed to get session %s: %s", session_id, e)
    return None

  session = self._response_to_session(
      response,
      app_name=app_name,
      user_id=user_id,
      session_id=session_id,
  )

  if config:
    if config.num_recent_events is not None:
      session.events = session.events[-config.num_recent_events :]
    if config.after_timestamp is not None:
      session.events = [
          event
          for event in session.events
          if event.timestamp >= config.after_timestamp
      ]

  return session

list_sessions async

list_sessions(*, app_name: str, user_id: str | None = None) -> ListSessionsResponse

List stored sessions from Redis Agent Memory.

Source code in src/adk_redis/sessions/session_memory.py

@override
async def list_sessions(
    self, *, app_name: str, user_id: str | None = None
) -> ListSessionsResponse:
  """List stored sessions from Redis Agent Memory."""
  if self._config.backend == OPENSOURCE_AGENT_MEMORY_BACKEND:
    return await self._list_sessions_agent_memory_server(
        app_name=app_name,
        user_id=user_id,
    )

  namespace = self._get_namespace(app_name)
  sessions = []
  page_token = None

  try:
    async with self._agent_memory() as agent_memory:
      while True:
        response = await agent_memory.list_sessions_async(
            limit=100,
            page_token=page_token,
        )
        for storage_session_id in read_field(response, "items", []) or []:
          parsed = self._parse_storage_session_id(storage_session_id)
          if parsed:
            stored_namespace, stored_user_id, session_id = parsed
          else:
            try:
              stored = await agent_memory.get_session_memory_async(
                  session_id=storage_session_id
              )
            except Exception:
              continue
            scope = self._session_scope_from_response(stored)
            if not scope:
              continue
            stored_namespace, stored_user_id, session_id = scope
          if stored_namespace != namespace:
            continue
          if user_id is not None and stored_user_id != user_id:
            continue
          sessions.append(
              Session(
                  id=session_id,
                  app_name=app_name,
                  user_id=stored_user_id,
                  state={},
                  events=[],
                  last_update_time=time.time(),
              )
          )

        page_token = read_field(response, "next_page_token")
        if not page_token:
          break

    return ListSessionsResponse(sessions=sessions)

  except Exception as e:
    logger.error("Failed to list sessions: %s", e)
    return ListSessionsResponse(sessions=[])

delete_session async

delete_session(*, app_name: str, user_id: str, session_id: str) -> None

Delete a session from Redis Agent Memory.

Source code in src/adk_redis/sessions/session_memory.py

@override
async def delete_session(
    self, *, app_name: str, user_id: str, session_id: str
) -> None:
  """Delete a session from Redis Agent Memory."""
  if self._config.backend == OPENSOURCE_AGENT_MEMORY_BACKEND:
    await self._delete_session_agent_memory_server(
        app_name=app_name,
        user_id=user_id,
        session_id=session_id,
    )
    return

  storage_session_id = self._storage_session_id(
      app_name=app_name,
      user_id=user_id,
      session_id=session_id,
  )
  try:
    async with self._agent_memory() as agent_memory:
      await agent_memory.delete_session_memory_async(
          session_id=storage_session_id
      )
    logger.info("Deleted session %s", session_id)
  except Exception as e:
    if not is_not_found_error(e):
      logger.error("Failed to delete session %s: %s", session_id, e)

append_event async

append_event(session: Session, event: Event) -> Event

Append an event to the ADK session and configured backend.

Source code in src/adk_redis/sessions/session_memory.py

@override
async def append_event(self, session: Session, event: Event) -> Event:
  """Append an event to the ADK session and configured backend."""
  appended_event = await super().append_event(session=session, event=event)
  if appended_event.partial:
    return appended_event

  session.last_update_time = appended_event.timestamp
  try:
    if self._config.backend == OPENSOURCE_AGENT_MEMORY_BACKEND:
      await self._append_event_to_agent_memory_server(
          session,
          appended_event,
      )
    else:
      await self._append_event_to_redis(session, appended_event)
    logger.debug("Appended message to session %s", session.id)
  except Exception as e:
    logger.error("Failed to append event to session %s: %s", session.id, e)

  return appended_event

close async

close() -> None

Close the session service and cleanup resources.

Source code in src/adk_redis/sessions/session_memory.py

async def close(self) -> None:
  """Close the session service and cleanup resources."""
  pass

RedisSessionMemoryServiceConfig

Bases: BaseModel

Configuration for Redis session memory service.

Attributes:

Name	Type	Description
backend	MemoryBackendName	Memory backend to use.
api_base_url	str	Memory API base URL.
api_key	str | None	Redis Agent Memory API key.
store_id	str | None	Redis Agent Memory store ID.
timeout	float	HTTP request timeout in seconds.
timeout_ms	int | None	Optional SDK timeout in milliseconds.
default_namespace	str | None	Default namespace for session isolation.
model_name	str | None	Backward-compatible option retained from the previous client.
context_window_max	int | None	Backward-compatible option retained from the previous client.
extraction_strategy	Literal['discrete', 'summary', 'preferences', 'custom']	Backward-compatible option retained from the previous client.
extraction_strategy_config	dict[str, Any]	Backward-compatible option retained from the previous client.
session_ttl_seconds	int | None	Backward-compatible option retained from the previous client.

RedisWorkingMemorySessionService

RedisWorkingMemorySessionService(config: RedisSessionMemoryServiceConfig | None = None)

Bases: RedisSessionMemoryService

Deprecated alias for :class:RedisSessionMemoryService.

Retained for backward compatibility and removed in 0.1.0.

Source code in src/adk_redis/sessions/session_memory.py

def __init__(self, config: RedisSessionMemoryServiceConfig | None = None):
  warnings.warn(
      "RedisWorkingMemorySessionService is deprecated; use "
      "RedisSessionMemoryService. The alias will be removed in 0.1.0.",
      DeprecationWarning,
      stacklevel=2,
  )
  super().__init__(config=config)

create_session async

create_session(*, app_name: str, user_id: str, state: dict[str, Any] | None = None, session_id: str | None = None) -> Session

Create a new ADK session object.

Redis Agent Memory creates a stored session when the first event is appended, so this method returns the ADK session without writing a marker event.

Source code in src/adk_redis/sessions/session_memory.py

@override
async def create_session(
    self,
    *,
    app_name: str,
    user_id: str,
    state: dict[str, Any] | None = None,
    session_id: str | None = None,
) -> Session:
  """Create a new ADK session object.

  Redis Agent Memory creates a stored session when the first event is
  appended, so this method returns the ADK session without writing a marker
  event.
  """
  if self._config.backend == OPENSOURCE_AGENT_MEMORY_BACKEND:
    return await self._create_session_agent_memory_server(
        app_name=app_name,
        user_id=user_id,
        state=state,
        session_id=session_id,
    )

  session_id = (
      session_id.strip()
      if session_id and session_id.strip()
      else str(uuid.uuid4())
  )
  return Session(
      id=session_id,
      app_name=app_name,
      user_id=user_id,
      state=state or {},
      events=[],
      last_update_time=time.time(),
  )

get_session async

get_session(*, app_name: str, user_id: str, session_id: str, config: GetSessionConfig | None = None) -> Session | None

Retrieve an ADK session from Redis Agent Memory.

Source code in src/adk_redis/sessions/session_memory.py

@override
async def get_session(
    self,
    *,
    app_name: str,
    user_id: str,
    session_id: str,
    config: GetSessionConfig | None = None,
) -> Session | None:
  """Retrieve an ADK session from Redis Agent Memory."""
  if self._config.backend == OPENSOURCE_AGENT_MEMORY_BACKEND:
    return await self._get_session_agent_memory_server(
        app_name=app_name,
        user_id=user_id,
        session_id=session_id,
        config=config,
    )

  storage_session_id = self._storage_session_id(
      app_name=app_name,
      user_id=user_id,
      session_id=session_id,
  )

  try:
    async with self._agent_memory() as agent_memory:
      response = await agent_memory.get_session_memory_async(
          session_id=storage_session_id
      )
  except Exception as e:
    if is_not_found_error(e):
      # Managed sessions are materialized on first append_event. Return an
      # empty ADK session so runners can start a new conversation.
      return Session(
          id=session_id,
          app_name=app_name,
          user_id=user_id,
          state={},
          events=[],
          last_update_time=time.time(),
      )
    logger.error("Failed to get session %s: %s", session_id, e)
    return None

  session = self._response_to_session(
      response,
      app_name=app_name,
      user_id=user_id,
      session_id=session_id,
  )

  if config:
    if config.num_recent_events is not None:
      session.events = session.events[-config.num_recent_events :]
    if config.after_timestamp is not None:
      session.events = [
          event
          for event in session.events
          if event.timestamp >= config.after_timestamp
      ]

  return session

list_sessions async

list_sessions(*, app_name: str, user_id: str | None = None) -> ListSessionsResponse

List stored sessions from Redis Agent Memory.

Source code in src/adk_redis/sessions/session_memory.py

@override
async def list_sessions(
    self, *, app_name: str, user_id: str | None = None
) -> ListSessionsResponse:
  """List stored sessions from Redis Agent Memory."""
  if self._config.backend == OPENSOURCE_AGENT_MEMORY_BACKEND:
    return await self._list_sessions_agent_memory_server(
        app_name=app_name,
        user_id=user_id,
    )

  namespace = self._get_namespace(app_name)
  sessions = []
  page_token = None

  try:
    async with self._agent_memory() as agent_memory:
      while True:
        response = await agent_memory.list_sessions_async(
            limit=100,
            page_token=page_token,
        )
        for storage_session_id in read_field(response, "items", []) or []:
          parsed = self._parse_storage_session_id(storage_session_id)
          if parsed:
            stored_namespace, stored_user_id, session_id = parsed
          else:
            try:
              stored = await agent_memory.get_session_memory_async(
                  session_id=storage_session_id
              )
            except Exception:
              continue
            scope = self._session_scope_from_response(stored)
            if not scope:
              continue
            stored_namespace, stored_user_id, session_id = scope
          if stored_namespace != namespace:
            continue
          if user_id is not None and stored_user_id != user_id:
            continue
          sessions.append(
              Session(
                  id=session_id,
                  app_name=app_name,
                  user_id=stored_user_id,
                  state={},
                  events=[],
                  last_update_time=time.time(),
              )
          )

        page_token = read_field(response, "next_page_token")
        if not page_token:
          break

    return ListSessionsResponse(sessions=sessions)

  except Exception as e:
    logger.error("Failed to list sessions: %s", e)
    return ListSessionsResponse(sessions=[])

delete_session async

delete_session(*, app_name: str, user_id: str, session_id: str) -> None

Delete a session from Redis Agent Memory.

Source code in src/adk_redis/sessions/session_memory.py

@override
async def delete_session(
    self, *, app_name: str, user_id: str, session_id: str
) -> None:
  """Delete a session from Redis Agent Memory."""
  if self._config.backend == OPENSOURCE_AGENT_MEMORY_BACKEND:
    await self._delete_session_agent_memory_server(
        app_name=app_name,
        user_id=user_id,
        session_id=session_id,
    )
    return

  storage_session_id = self._storage_session_id(
      app_name=app_name,
      user_id=user_id,
      session_id=session_id,
  )
  try:
    async with self._agent_memory() as agent_memory:
      await agent_memory.delete_session_memory_async(
          session_id=storage_session_id
      )
    logger.info("Deleted session %s", session_id)
  except Exception as e:
    if not is_not_found_error(e):
      logger.error("Failed to delete session %s: %s", session_id, e)

append_event async

append_event(session: Session, event: Event) -> Event

Append an event to the ADK session and configured backend.

Source code in src/adk_redis/sessions/session_memory.py

@override
async def append_event(self, session: Session, event: Event) -> Event:
  """Append an event to the ADK session and configured backend."""
  appended_event = await super().append_event(session=session, event=event)
  if appended_event.partial:
    return appended_event

  session.last_update_time = appended_event.timestamp
  try:
    if self._config.backend == OPENSOURCE_AGENT_MEMORY_BACKEND:
      await self._append_event_to_agent_memory_server(
          session,
          appended_event,
      )
    else:
      await self._append_event_to_redis(session, appended_event)
    logger.debug("Appended message to session %s", session.id)
  except Exception as e:
    logger.error("Failed to append event to session %s: %s", session.id, e)

  return appended_event

close async

close() -> None

Close the session service and cleanup resources.

Source code in src/adk_redis/sessions/session_memory.py

async def close(self) -> None:
  """Close the session service and cleanup resources."""
  pass

RedisWorkingMemorySessionServiceConfig

RedisWorkingMemorySessionServiceConfig(**data: Any)

Bases: RedisSessionMemoryServiceConfig

Deprecated alias for :class:RedisSessionMemoryServiceConfig.

Retained for backward compatibility and removed in 0.1.0.

Source code in src/adk_redis/sessions/session_memory.py

def __init__(self, **data: Any):
  warnings.warn(
      "RedisWorkingMemorySessionServiceConfig is deprecated; use "
      "RedisSessionMemoryServiceConfig. The alias will be removed in 0.1.0.",
      DeprecationWarning,
      stacklevel=2,
  )
  super().__init__(**data)