Google ADK

Give an agent built with Google's Agent Development Kit a memory of each of your users. Geniffy plugs in as the runner's memory service, so ADK's own memory tools use it: preload_memory puts what is known in front of the model on every turn, and load_memory lets the model look things up when it decides to. Each ADK user is a space of their own.

Install

Terminal

pip install google-adk geniffy

Terminal

uv add google-adk geniffy

Set GOOGLE_API_KEY, and GENIFFY_API_KEY from API keys in the Geniffy app.

The memory service

geniffy_memory.py

from geniffy import AsyncGeniffy
from google.adk.events import Event
from google.adk.memory import BaseMemoryService
from google.adk.memory.base_memory_service import SearchMemoryResponse
from google.adk.memory.memory_entry import MemoryEntry
from google.adk.sessions import Session
from google.genai import types

geniffy = AsyncGeniffy()                 # reads GENIFFY_API_KEY


def text_of(event: Event) -> str:
    parts = event.content.parts if event.content and event.content.parts else []
    return "".join(part.text or "" for part in parts)


def message_of(event: Event) -> dict:
    role = "user" if event.author == "user" else "assistant"
    return {"role": role, "content": text_of(event)}


class GeniffyMemoryService(BaseMemoryService):
    """Each ADK user is a Geniffy space: search reads it, and sessions are added to it."""

    async def search_memory(
        self, *, app_name: str, user_id: str, query: str,
    ) -> SearchMemoryResponse:
        known = await geniffy.space(f"user_{user_id}").context(query)
        entry = MemoryEntry(content=types.Content(parts=[types.Part(text=known)]))
        return SearchMemoryResponse(memories=[entry])

    async def add_session_to_memory(self, session: Session) -> None:
        messages = [message_of(event) for event in session.events if text_of(event)]
        if messages:
            await geniffy.space(f"user_{session.user_id}").memories.add(messages=messages)

search_memory answers with context(): what is known that bears on the query, each line with where it came from, or one sentence when nothing is known, so the model doesn't guess. add_session_to_memory sends the session as a conversation, and Geniffy keeps who said what.

Remember each user

Give the agent preload_memory, and add the session to memory when the conversation is over.

agent.py

from geniffy_memory import GeniffyMemoryService
from google.adk.agents import Agent
from google.adk.runners import Runner
from google.adk.sessions import InMemorySessionService
from google.adk.tools import preload_memory
from google.genai import types

APP = "assistant"
agent = Agent(name="assistant", model="gemini-flash-latest",
              instruction="You are a helpful assistant.", tools=[preload_memory])
runner = Runner(agent=agent, app_name=APP, memory_service=GeniffyMemoryService(),
                session_service=InMemorySessionService())


async def chat(user_id: str, message: str) -> str:
    sessions = runner.session_service
    session = await sessions.create_session(app_name=APP, user_id=user_id)
    said = types.Content(role="user", parts=[types.Part(text=message)])
    reply = ""
    events = runner.run_async(user_id=user_id, session_id=session.id, new_message=said)
    async for event in events:
        if event.is_final_response() and event.content and event.content.parts:
            reply = "".join(part.text or "" for part in event.content.parts)

    done = await sessions.get_session(app_name=APP, user_id=user_id, session_id=session.id)
    await runner.memory_service.add_session_to_memory(done)
    return reply

Call it with the user from your own sign-in:

reply = await chat(user.id, "Who signs the Lumen renewal?")

Here each message is its own session, so it is added at once. In a longer conversation, add the session when it ends, so each turn reaches Geniffy once.

Let the agent look things up

To let the agent decide when to look something up, give it load_memory instead. ADK calls the same memory service with the user of the session, so the model never sees or chooses whose memory it reads.

agent.py

from geniffy_memory import GeniffyMemoryService
from google.adk.agents import Agent
from google.adk.runners import Runner
from google.adk.sessions import InMemorySessionService
from google.adk.tools import load_memory
from google.genai import types

APP = "assistant"
INSTRUCTION = ("You are a helpful assistant. Use load_memory before answering anything "
               "that depends on what the user said before.")
agent = Agent(name="assistant", model="gemini-flash-latest",
              instruction=INSTRUCTION, tools=[load_memory])
runner = Runner(agent=agent, app_name=APP, memory_service=GeniffyMemoryService(),
                session_service=InMemorySessionService())


async def chat(user_id: str, message: str) -> str:
    sessions = runner.session_service
    session = await sessions.create_session(app_name=APP, user_id=user_id)
    said = types.Content(role="user", parts=[types.Part(text=message)])
    reply = ""
    events = runner.run_async(user_id=user_id, session_id=session.id, new_message=said)
    async for event in events:
        if event.is_final_response() and event.content and event.content.parts:
            reply = "".join(part.text or "" for part in event.content.parts)

    done = await sessions.get_session(app_name=APP, user_id=user_id, session_id=session.id)
    await runner.memory_service.add_session_to_memory(done)
    return reply

Last updated October 5, 2026