mem0 Memory System
"""
Memory provider for the mem0 MCP server
"""
import os
import sys
import json
import logging
from typing import Dict, Any, List, Optional, Union
# Set DOCS_MODE for documentation
os.environ["DOCS_MODE"] = "1"
# Try to import mem0 or mem0ai
try:
try:
# First try to import from mem0ai (newer package name)
import mem0ai
from mem0ai.memory import Memory as Mem0
MEM0_PACKAGE = "mem0ai"
except (ImportError, AttributeError):
# Fall back to mem0 (older package name)
import mem0
from mem0.memory import Memory as Mem0
MEM0_PACKAGE = "mem0"
except (ImportError, AttributeError):
# Create a mock Mem0 class for documentation purposes
class Mem0:
def __init__(self, *args, **kwargs):
print(f"Mock Mem0 initialized with args: {args}, kwargs: {kwargs}")
self.memories = {}
def add(self, content, metadata=None, **kwargs):
memory_id = f"memory-{len(self.memories) + 1}"
self.memories[memory_id] = {"content": content, "metadata": metadata or {}}
print(f"Mock memory added with ID: {memory_id}")
return memory_id
def search(self, query, limit=5, **kwargs):
print(f"Mock search for query: {query}, limit: {limit}")
results = []
for memory_id, memory in self.memories.items():
results.append({
"id": memory_id,
"content": memory["content"],
"metadata": memory["metadata"],
"score": 0.9 # Mock relevance score
})
return results[:limit]
def get(self, memory_id, **kwargs):
print(f"Mock get memory with ID: {memory_id}")
return self.memories.get(memory_id, {})
def delete(self, memory_id, **kwargs):
print(f"Mock delete memory with ID: {memory_id}")
if memory_id in self.memories:
del self.memories[memory_id]
def clear(self, **kwargs):
print("Mock clear all memories")
self.memories = {}
# Only raise if we're not in documentation mode
if not os.environ.get("DOCS_MODE"):
raise ImportError(
"Could not import mem0 or mem0ai package. Please run the installer script: bash install.sh"
)
MEM0_PACKAGE = None
# Set DOCS_MODE for documentation
os.environ["DOCS_MODE"] = "1"
class MemoryProvider:
"""
Memory provider for the mem0 MCP server
"""
# Available LLM providers
PROVIDERS = [
"openai",
"anthropic",
"google",
"ollama",
"deepseek",
"together",
"groq",
"bedrock",
"azure",
"litellm"
]
# Available embedding providers
EMBEDDING_PROVIDERS = [
"openai",
"huggingface",
"ollama",
"together",
"google",
"azure"
]
def __init__(
self,
provider: str = "openai",
embedding_provider: str = "openai",
api_key: Optional[str] = None,
model: Optional[str] = None,
embedding_model: Optional[str] = None,
data_dir: str = "./memory_data",
):
"""
Initialize the memory provider
Args:
provider: LLM provider to use
embedding_provider: Embedding provider to use
api_key: API key for the provider
model: Model to use for the provider
embedding_model: Model to use for embeddings
data_dir: Directory to store memory data
"""
# Set environment variables for API keys if provided
if api_key:
if provider == "openai":
os.environ["OPENAI_API_KEY"] = api_key
elif provider == "anthropic":
os.environ["ANTHROPIC_API_KEY"] = api_key
elif provider == "google":
os.environ["GOOGLE_API_KEY"] = api_key
# Create data directory if it doesn't exist
os.makedirs(data_dir, exist_ok=True)
# Initialize mem0 client
self.mem0 = Mem0(
provider=provider,
embedding_provider=embedding_provider,
model=model,
embedding_model=embedding_model,
data_dir=data_dir,
)
# Store configuration
self.config = {
"provider": provider,
"embedding_provider": embedding_provider,
"model": model,
"embedding_model": embedding_model,
"data_dir": data_dir,
}
# Log initialization
logging.info(f"Memory provider initialized with {provider} and {embedding_provider}")
def store(self, content: str, metadata: Optional[Dict[str, Any]] = None) -> str:
"""
Store a memory
Args:
content: Content to store
metadata: Optional metadata
Returns:
Memory ID
"""
memory_id = self.mem0.add(content, metadata=metadata)
return memory_id
def retrieve(self, query: str, limit: int = 5) -> List[Dict[str, Any]]:
"""
Retrieve memories based on query
Args:
query: Query to search for
limit: Maximum number of results
Returns:
List of memories
"""
results = self.mem0.search(query, limit=limit)
return results
def get(self, memory_id: str) -> Dict[str, Any]:
"""
Get a memory by ID
Args:
memory_id: Memory ID
Returns:
Memory
"""
memory = self.mem0.get(memory_id)
return memory
def delete(self, memory_id: str) -> None:
"""
Delete a memory by ID
Args:
memory_id: Memory ID
"""
self.mem0.delete(memory_id)
def clear(self, user_id: Optional[str] = None) -> None:
"""
Clear all memories for a user or all users
Args:
user_id: User ID to clear memories for, or None for all users
"""
self.mem0.clear(user_id=user_id)
def list_all(self) -> List[Dict[str, Any]]:
"""
List all stored memories
Returns:
List of memories
"""
# This is a mock implementation as mem0 doesn't have a list_all method
# In a real implementation, you would use the mem0 API to list all memories
return []