mem0 Memory System
"""
Client library for the mem0 MCP server
This module provides a client for interacting with the mem0 MCP server.
"""
import requests
from typing import Dict, Any, Optional, List, Union
class Mem0Client:
"""Client for the mem0 MCP server"""
def __init__(self, base_url: str = "http://localhost:8000"):
"""
Initialize the client
Args:
base_url: Base URL of the mem0 MCP server
"""
self.base_url = base_url.rstrip("/")
def configure(
self,
provider: str = "openai",
data_dir: str = "./memory_data",
collection_name: str = "memories",
embedding_provider: str = "openai",
embedding_model: str = "text-embedding-3-small",
llm_model: Optional[str] = None,
temperature: float = 0.7,
max_tokens: int = 1024,
) -> Dict[str, Any]:
"""
Configure the memory provider
Args:
provider: LLM provider to use
data_dir: Directory to store memory data
collection_name: Name for the ChromaDB collection
embedding_provider: Provider to use for embeddings
embedding_model: Model to use for embeddings
llm_model: Model to use for LLM
temperature: Temperature for LLM generation
max_tokens: Maximum tokens for LLM generation
Returns:
Response from the server
"""
config = {
"provider": provider,
"data_dir": data_dir,
"collection_name": collection_name,
"embedding_provider": embedding_provider,
"embedding_model": embedding_model,
"temperature": temperature,
"max_tokens": max_tokens,
}
if llm_model:
config["llm_model"] = llm_model
response = requests.post(f"{self.base_url}/configure", json=config)
response.raise_for_status()
return response.json()
def add_memory(
self,
content: Union[str, List[Dict[str, str]]],
user_id: str = "default_user",
metadata: Optional[Dict[str, Any]] = None,
) -> str:
"""
Add a memory
Args:
content: String content or list of message dicts with 'role' and 'content'
user_id: User ID to associate with the memory
metadata: Additional metadata for the memory
Returns:
Memory ID
"""
data = {
"content": content,
"user_id": user_id,
}
if metadata:
data["metadata"] = metadata
response = requests.post(f"{self.base_url}/memory/add", json=data)
response.raise_for_status()
return response.json()["memory_id"]
def search_memories(
self,
query: str,
user_id: str = "default_user",
limit: int = 5,
metadata_filter: Optional[Dict[str, Any]] = None,
) -> Dict[str, Any]:
"""
Search for memories
Args:
query: Search query
user_id: User ID to search memories for
limit: Maximum number of results to return
metadata_filter: Filter results by metadata
Returns:
Search results with scores
"""
data = {
"query": query,
"user_id": user_id,
"limit": limit,
}
if metadata_filter:
data["metadata_filter"] = metadata_filter
response = requests.post(f"{self.base_url}/memory/search", json=data)
response.raise_for_status()
return response.json()
def chat(
self,
message: str,
user_id: str = "default_user",
system_prompt: Optional[str] = None,
include_memories: bool = True,
) -> str:
"""
Chat with the AI using memories
The system automatically:
1. Extracts and stores user information from the message
2. Retrieves relevant memories based on the message
3. Enhances the system prompt with this information
4. Returns a response that naturally incorporates user information
Args:
message: User message
user_id: User ID for memory retrieval
system_prompt: Custom system prompt (optional)
include_memories: Whether to include memories in the prompt
Returns:
AI response
"""
data = {
"message": message,
"user_id": user_id,
"include_memories": include_memories,
}
if system_prompt:
data["system_prompt"] = system_prompt
response = requests.post(f"{self.base_url}/memory/chat", json=data)
response.raise_for_status()
return response.json()["response"]
def get_memory(self, memory_id: str) -> Dict[str, Any]:
"""
Get a memory by ID
Args:
memory_id: Memory ID
Returns:
Memory data
"""
response = requests.get(f"{self.base_url}/memory/{memory_id}")
response.raise_for_status()
return response.json()
def delete_memory(self, memory_id: str) -> bool:
"""
Delete a memory by ID
Args:
memory_id: Memory ID
Returns:
True if deleted
"""
response = requests.delete(f"{self.base_url}/memory/{memory_id}")
response.raise_for_status()
return response.json()["success"]
def clear_memories(self, user_id: Optional[str] = None) -> bool:
"""
Clear all memories for a user or all users
Args:
user_id: User ID to clear memories for (if None, clears all)
Returns:
True if cleared
"""
params = {}
if user_id:
params["user_id"] = user_id
response = requests.delete(f"{self.base_url}/memories", params=params)
response.raise_for_status()
return response.json()["success"]
def health_check(self) -> bool:
"""
Check if the server is running
Returns:
True if the server is running
"""
try:
response = requests.get(f"{self.base_url}/health")
response.raise_for_status()
return response.json()["success"]
except:
return False
def list_providers(self) -> Dict[str, Any]:
"""
List available providers
Returns:
Dictionary of available providers
"""
response = requests.get(f"{self.base_url}/providers")
response.raise_for_status()
return response.json()