Google ADK - Mem0

Integrate Mem0 with Google ADK (Agent Development Kit), an open-source framework for building multi-agent workflows. This integration enables agents to access persistent memory across conversations, enhancing context retention and personalization.

Overview

In this guide, we’ll create a Google ADK agent that:

Uses ADK’s native MemoryService interface to connect Mem0
Automatically injects relevant memories using ADK’s built-in load_memory tool
Persists session history to Mem0 after each turn via an after-agent callback
Shares memory seamlessly across multi-agent hierarchies

Setup and Configuration

Install the necessary libraries:

pip install google-adk mem0ai python-dotenv

Set up your API keys:

Mem0 API Key
Google AI Studio API Key

Remember to get your API key from Mem0 Platform and set up a Google AI Studio API Key.

import os
from dotenv import load_dotenv

load_dotenv()

# os.environ["GOOGLE_API_KEY"] = "your-google-api-key"
# os.environ["MEM0_API_KEY"] = "your-mem0-api-key"

Implement Mem0MemoryService

Create a custom MemoryService by implementing ADK’s BaseMemoryService. Save the following as mem0_memory_service.py:

import asyncio
import os
from typing import Optional
from typing_extensions import override

from google.adk.memory.base_memory_service import BaseMemoryService, SearchMemoryResponse
from google.adk.memory.memory_entry import MemoryEntry
from google.adk.sessions import Session
from google.genai.types import Content, Part
from mem0 import MemoryClient


class Mem0MemoryService(BaseMemoryService):
    """MemoryService implementation backed by the Mem0 Platform."""

    def __init__(self, api_key: Optional[str] = None):
        super().__init__()
        api_key = api_key or os.environ.get("MEM0_API_KEY")
        self._client: Optional[MemoryClient] = MemoryClient(api_key=api_key) if api_key else None

    @override
    async def search_memory(
        self, *, app_name: str, user_id: str, query: str
    ) -> SearchMemoryResponse:
        """Search for memories relevant to the current user and query."""
        if not self._client:
            return SearchMemoryResponse(memories=[])

        try:
            results = await asyncio.to_thread(
                self._client.search,
                query,
                filters={"AND": [{"user_id": user_id}, {"app_id": app_name}]},
                top_k=5,
            )

            entries = []
            for mem in results.get("results", []):
                text = mem.get("memory", "")
                if not text:
                    continue

                raw_ts = mem.get("created_at") or mem.get("updated_at")
                entries.append(
                    MemoryEntry(
                        content=Content(parts=[Part(text=text)]),
                        author=mem.get("metadata", {}).get("author", "user"),
                        timestamp=str(raw_ts) if raw_ts else None,
                    )
                )

            return SearchMemoryResponse(memories=entries)

        except Exception as e:
            print(f"[Mem0MemoryService] search_memory error: {e}")
            return SearchMemoryResponse(memories=[])

    @override
    async def add_session_to_memory(self, session: Session) -> None:
        """Persist a completed ADK session into Mem0."""
        if not self._client:
            return

        user_id = session.user_id
        if not user_id:
            return

        app_name = getattr(session, "app_name", None)

        try:
            messages = []
            for event in session.events:
                if not (event.content and event.content.parts):
                    continue
                role = getattr(event.content, "role", None) or "user"
                if role == "model":
                    role = "assistant"
                elif role not in ("user", "assistant"):
                    continue
                text_parts = [
                    p.text for p in event.content.parts if hasattr(p, "text") and p.text
                ]
                if text_parts:
                    messages.append({"role": role, "content": " ".join(text_parts)})

            if messages:
                metadata = {"app_id": app_name} if app_name else {}
                await asyncio.to_thread(
                    self._client.add, messages, user_id=user_id, metadata=metadata
                )

        except Exception as e:
            print(f"[Mem0MemoryService] add_session_to_memory error: {e}")

Add Auto-Save Callback

This after-agent callback fires at the end of every turn and saves the session to Mem0. Save as memory_callbacks.py:

async def save_session_to_memory(callback_context) -> None:
    """Persist the completed session to Mem0 after each agent turn."""
    try:
        await callback_context.add_session_to_memory()
    except ValueError:
        pass
    except Exception as e:
        print(f"[save_session_to_memory] error: {e}")

Basic Integration Example

The following example demonstrates creating an ADK agent with automatic Mem0 memory:

import asyncio
from google.adk.agents import LlmAgent
from google.adk.runners import Runner
from google.adk.sessions import InMemorySessionService
from google.adk.tools import load_memory
from google.genai.types import Content, Part

from mem0_memory_service import Mem0MemoryService
from memory_callbacks import save_session_to_memory

memory_service = Mem0MemoryService()
session_service = InMemorySessionService()

agent = LlmAgent(
    name="personal_assistant",
    model="gemini-2.0-flash",
    instruction="""You are a helpful personal assistant.
    Relevant memories from past conversations are provided to you automatically.
    Use them to personalize your responses.""",
    description="A personal assistant that remembers user preferences and past interactions",
    tools=[load_memory],
    after_agent_callback=save_session_to_memory,
)

runner = Runner(
    agent=agent,
    session_service=session_service,
    memory_service=memory_service,
    app_name="memory_assistant",
)


async def chat(user_input: str, user_id: str) -> str:
    session = await session_service.create_session(
        app_name="memory_assistant",
        user_id=user_id,
    )
    content = Content(role="user", parts=[Part(text=user_input)])
    async for event in runner.run_async(user_id=user_id, session_id=session.id, new_message=content):
        if event.is_final_response() and event.content and event.content.parts:
            return event.content.parts[0].text
    return "No response generated"


if __name__ == "__main__":
    print(asyncio.run(chat(
        "I love Italian food and I'm planning a trip to Rome next month",
        user_id="alice",
    )))

    print(asyncio.run(chat(
        "Any food recommendations for my trip?",
        user_id="alice",
    )))

Multi-Agent Hierarchy with Shared Memory

Because memory_service is passed to the Runner, every agent in the hierarchy shares the same memory automatically. Only the root coordinator needs the auto-save callback — ADK fires it once when the full turn completes:

import asyncio
from google.adk.agents import LlmAgent
from google.adk.runners import Runner
from google.adk.sessions import InMemorySessionService
from google.adk.tools.agent_tool import AgentTool
from google.adk.tools import load_memory
from google.genai.types import Content, Part

from mem0_memory_service import Mem0MemoryService
from memory_callbacks import save_session_to_memory

memory_service = Mem0MemoryService()
session_service = InMemorySessionService()

travel_agent = LlmAgent(
    name="travel_specialist",
    model="gemini-2.0-flash",
    instruction="""You are a travel planning specialist.
    Relevant memories about the user's travel preferences are provided automatically.
    Use them to make personalized recommendations.""",
    description="Specialist in travel planning and recommendations",
    tools=[load_memory],
)

health_agent = LlmAgent(
    name="health_advisor",
    model="gemini-2.0-flash",
    instruction="""You are a health and wellness advisor.
    Relevant memories about the user's health goals are provided automatically.
    Use them to give personalized advice.""",
    description="Specialist in health and wellness advice",
    tools=[load_memory],
)

coordinator = LlmAgent(
    name="coordinator",
    model="gemini-2.0-flash",
    instruction="""You are a coordinator that delegates requests to specialist agents.
    For travel-related questions, delegate to the travel specialist.
    For health-related questions, delegate to the health advisor.
    Relevant memories about the user are provided automatically.""",
    description="Coordinates requests between specialist agents",
    tools=[
        load_memory,
        AgentTool(agent=travel_agent, skip_summarization=False),
        AgentTool(agent=health_agent, skip_summarization=False),
    ],
    after_agent_callback=save_session_to_memory,
)

runner = Runner(
    agent=coordinator,
    session_service=session_service,
    memory_service=memory_service,
    app_name="specialist_system",
)


async def chat_with_specialists(user_input: str, user_id: str) -> str:
    session = await session_service.create_session(
        app_name="specialist_system",
        user_id=user_id,
    )
    content = Content(role="user", parts=[Part(text=user_input)])
    async for event in runner.run_async(user_id=user_id, session_id=session.id, new_message=content):
        if event.is_final_response() and event.content and event.content.parts:
            return event.content.parts[0].text
    return "No response generated"


if __name__ == "__main__":
    response = asyncio.run(chat_with_specialists("Plan a healthy meal for my Italy trip", user_id="alice"))
    print(response)

Key Features

Automatic Memory Injection: ADK’s built-in load_memory tool searches Mem0 at the start of each turn and injects relevant memories directly into the agent context — no prompt instructions needed.
Automatic Session Saving: The save_session_to_memory callback persists every completed turn to Mem0 without any manual calls.
Native ADK Integration: Mem0MemoryService implements ADK’s BaseMemoryService and integrates via the Runner — works natively across the entire agent hierarchy.
User Scoping: user_id is passed automatically from the ADK session context, ensuring memories are always scoped to the correct user.
Multi-Agent Support: A single Mem0MemoryService instance shared through the Runner gives all agents — coordinators and specialists — access to the same user memory.

Configuration Options

Using Vertex AI

To use Google Cloud Vertex AI instead of AI Studio, set the following environment variables before creating agents:

import os
os.environ["GOOGLE_GENAI_USE_VERTEXAI"] = "True"
os.environ["GOOGLE_CLOUD_PROJECT"] = "your-project-id"
os.environ["GOOGLE_CLOUD_LOCATION"] = "us-central1"

Advanced Memory Filtering

You can customize how memories are searched by modifying Mem0MemoryService.search_memory. For example, to filter by category:

results = await asyncio.to_thread(
    self._client.search,
    query,
    filters={
        "AND": [
            {"user_id": user_id},
            {"app_id": app_name},
            {"categories": {"contains": "travel"}}
        ]
    },
    top_k=10,
)

InMemorySessionService stores sessions in memory and is intended for prototyping. For production, use a persistent session service and clean up sessions when they are no longer needed.

Conclusion

By implementing Mem0MemoryService as an ADK BaseMemoryService, you get persistent, user-scoped memory across single agents and complex multi-agent hierarchies with minimal code. Memory injection and session saving happen automatically, keeping your agent prompts clean and your token usage efficient.

Healthcare Agent Cookbook

Build HIPAA-compliant healthcare agents with Google ADK

OpenAI Agents SDK

Compare with OpenAI’s agent framework

​Overview

​Setup and Configuration

​Implement Mem0MemoryService

​Add Auto-Save Callback

​Basic Integration Example

​Multi-Agent Hierarchy with Shared Memory

​Key Features

​Configuration Options

​Using Vertex AI

​Advanced Memory Filtering

​Conclusion

Healthcare Agent Cookbook

OpenAI Agents SDK

Overview

Setup and Configuration

Implement Mem0MemoryService

Add Auto-Save Callback

Basic Integration Example

Multi-Agent Hierarchy with Shared Memory

Key Features

Configuration Options

Using Vertex AI

Advanced Memory Filtering

Conclusion