---

name: mem0-fastapi-integration description: Memory layer integration patterns for FastAPI with Mem0 including client setup, memory service patterns, user tracking, conversation persistence, and background task integration. Use when implementing AI memory, adding Mem0 to FastAPI, building chat with memory, or when user mentions Mem0, conversation history, user context, or memory layer. allowed-tools: Read, Write, Bash, Grep, Glob

Mem0 FastAPI Integration Patterns

Purpose: Provide complete Mem0 integration templates, memory service patterns, user tracking implementations, and conversation persistence strategies for building FastAPI applications with intelligent AI memory.

Activation Triggers:

• Integrating Mem0 memory layer into FastAPI
• Building chat applications with conversation history
• Implementing user context and personalization
• Adding memory to AI agents
• Creating stateful AI interactions
• User preference management

Key Resources:

• templates/memory_service.py - Complete Mem0 service implementation
• templates/memory_middleware.py - Request-scoped memory middleware
• templates/memory_client.py - Mem0 client configuration
• templates/memory_routes.py - API routes for memory operations
• scripts/setup-mem0.sh - Mem0 installation and configuration
• scripts/test-memory.sh - Memory service testing utility
• examples/chat_with_memory.py - Complete chat implementation
• examples/user_preferences.py - User preference management

Core Mem0 Integration

1. Client Configuration

Template: templates/memory_client.py

Workflow:

from mem0 import Memory, AsyncMemory, MemoryClient
from mem0.configs.base import MemoryConfig

# Hosted Mem0 Platform
client = MemoryClient(api_key=settings.MEM0_API_KEY)

# Self-Hosted Configuration
config = MemoryConfig(
    vector_store={
        "provider": "qdrant",
        "config": {
            "host": settings.QDRANT_HOST,
            "port": settings.QDRANT_PORT,
            "api_key": settings.QDRANT_API_KEY
        }
    },
    llm={
        "provider": "openai",
        "config": {
            "model": "gpt-4",
            "api_key": settings.OPENAI_API_KEY
        }
    },
    embedder={
        "provider": "openai",
        "config": {
            "model": "text-embedding-3-small",
            "api_key": settings.OPENAI_API_KEY
        }
    }
)
memory = AsyncMemory(config)

2. Memory Service Pattern

Template: templates/memory_service.py

Key Operations:

class MemoryService:
    async def add_conversation(
        user_id: str,
        messages: List[Dict[str, str]],
        metadata: Optional[Dict] = None
    ) -> Dict

    async def search_memories(
        query: str,
        user_id: str,
        limit: int = 5
    ) -> List[Dict]

    async def get_user_summary(user_id: str) -> Dict

    async def add_user_preference(
        user_id: str,
        preference: str,
        category: str = "general"
    ) -> bool

Initialization:

@asynccontextmanager
async def lifespan(app: FastAPI):
    # Startup
    memory_service = MemoryService()
    app.state.memory_service = memory_service
    yield
    # Shutdown

Memory Patterns

1. Conversation Persistence

When to use: Chat applications, conversational AI

Template: templates/memory_service.py#add_conversation

async def add_conversation(
    self,
    user_id: str,
    messages: List[Dict[str, str]],
    metadata: Optional[Dict[str, Any]] = None
) -> Optional[Dict]:
    enhanced_metadata = {
        "timestamp": datetime.now().isoformat(),
        "conversation_type": "chat",
        **(metadata or {})
    }

    if self.client:
        result = self.client.add(
            messages=messages,
            user_id=user_id,
            metadata=enhanced_metadata
        )
    elif self.memory:
        result = await self.memory.add(
            messages=messages,
            user_id=user_id,
            metadata=enhanced_metadata
        )

    return result

Best for: Chat history, conversation context, multi-turn interactions

2. Semantic Memory Search

When to use: Context retrieval, relevant history lookup

Template: templates/memory_service.py#search_memories

async def search_memories(
    self,
    query: str,
    user_id: str,
    limit: int = 5,
    filters: Optional[Dict] = None
) -> List[Dict]:
    if self.client:
        result = self.client.search(
            query=query,
            user_id=user_id,
            limit=limit
        )
    elif self.memory:
        result = await self.memory.search(
            query=query,
            user_id=user_id,
            limit=limit
        )

    memories = result.get('results', [])
    return memories

Best for: Finding relevant past conversations, context-aware responses

3. User Preference Storage

When to use: Personalization, user settings, behavioral tracking

Template: templates/memory_service.py#add_user_preference

async def add_user_preference(
    self,
    user_id: str,
    preference: str,
    category: str = "general"
) -> bool:
    preference_message = {
        "role": "system",
        "content": f"User preference ({category}): {preference}"
    }

    metadata = {
        "type": "preference",
        "category": category,
        "timestamp": datetime.now().isoformat()
    }

    if self.client:
        self.client.add(
            messages=[preference_message],
            user_id=user_id,
            metadata=metadata
        )
    elif self.memory:
        await self.memory.add(
            messages=[preference_message],
            user_id=user_id,
            metadata=metadata
        )

    return True

Best for: User customization, learning user behavior, preference management

API Routes Integration

1. Memory Management Endpoints

Template: templates/memory_routes.py

from fastapi import APIRouter, Depends, BackgroundTasks
from app.api.deps import get_current_user, get_memory_service
from app.services.memory_service import MemoryService

router = APIRouter()

@router.post("/conversation")
async def add_conversation(
    request: ConversationRequest,
    background_tasks: BackgroundTasks,
    user_id: str = Depends(get_current_user),
    memory_service: MemoryService = Depends(get_memory_service)
):
    # Add to memory in background
    background_tasks.add_task(
        memory_service.add_conversation,
        user_id,
        request.messages,
        request.metadata
    )

    return {
        "status": "success",
        "message": "Conversation added to memory"
    }

@router.post("/search")
async def search_memories(
    request: SearchRequest,
    user_id: str = Depends(get_current_user),
    memory_service: MemoryService = Depends(get_memory_service)
):
    memories = await memory_service.search_memories(
        query=request.query,
        user_id=user_id,
        limit=request.limit
    )

    return {
        "query": request.query,
        "results": memories,
        "count": len(memories)
    }

@router.get("/summary")
async def get_memory_summary(
    user_id: str = Depends(get_current_user),
    memory_service: MemoryService = Depends(get_memory_service)
):
    summary = await memory_service.get_user_summary(user_id)
    return {"user_id": user_id, "summary": summary}

2. Request Models

Template: templates/memory_routes.py#models

from pydantic import BaseModel, Field
from typing import List, Dict, Optional, Any

class ConversationRequest(BaseModel):
    messages: List[Dict[str, str]] = Field(..., description="Conversation messages")
    session_id: Optional[str] = Field(None, description="Session identifier")
    metadata: Optional[Dict[str, Any]] = Field(None, description="Additional metadata")

class SearchRequest(BaseModel):
    query: str = Field(..., description="Search query")
    limit: int = Field(5, ge=1, le=20, description="Number of results")
    filters: Optional[Dict[str, Any]] = Field(None, description="Search filters")

class PreferenceRequest(BaseModel):
    preference: str = Field(..., description="User preference")
    category: str = Field("general", description="Preference category")

Background Task Integration

1. Async Memory Storage

Pattern:

from fastapi import BackgroundTasks

@router.post("/chat")
async def chat(
    request: ChatRequest,
    background_tasks: BackgroundTasks,
    memory_service: MemoryService = Depends(get_memory_service)
):
    # Generate AI response
    response = await ai_service.generate(request.message)

    # Store conversation in background
    background_tasks.add_task(
        memory_service.add_conversation,
        user_id=request.user_id,
        messages=[
            {"role": "user", "content": request.message},
            {"role": "assistant", "content": response}
        ]
    )

    return {"response": response}

Benefits:

• Non-blocking memory operations
• Faster response times
• Graceful failure handling

2. Batch Memory Updates

Pattern:

async def batch_update_memories(
    conversations: List[Conversation],
    memory_service: MemoryService
):
    tasks = [
        memory_service.add_conversation(
            user_id=conv.user_id,
            messages=conv.messages
        )
        for conv in conversations
    ]

    results = await asyncio.gather(*tasks, return_exceptions=True)
    return results

Dependency Injection Pattern

1. Memory Service Dependency

Template: templates/memory_middleware.py#deps

from fastapi import Request

def get_memory_service(request: Request) -> MemoryService:
    """Get memory service from app state"""
    return request.app.state.memory_service

# Usage in routes
@router.get("/")
async def endpoint(
    memory_service: MemoryService = Depends(get_memory_service)
):
    memories = await memory_service.search_memories(...)
    return memories

2. User Context Injection

Pattern:

async def get_user_context(
    user_id: str = Depends(get_current_user),
    memory_service: MemoryService = Depends(get_memory_service)
) -> Dict[str, Any]:
    """Get enriched user context from memory"""
    summary = await memory_service.get_user_summary(user_id)
    return {
        "user_id": user_id,
        "preferences": summary.get("user_preferences", []),
        "total_conversations": summary.get("total_memories", 0)
    }

# Usage
@router.post("/personalized-response")
async def personalized(
    request: QueryRequest,
    context: Dict = Depends(get_user_context)
):
    # Use context for personalized AI response
    pass

Implementation Workflow

Step 1: Setup Mem0

# Install and configure Mem0
./scripts/setup-mem0.sh

Creates:

• Mem0 configuration file
• Environment variable template
• Vector database setup (if self-hosted)

Step 2: Configure Memory Service

Decision tree:

• Hosted Mem0: Use MemoryClient with API key (simple, managed)
• Self-Hosted: Use AsyncMemory with config (full control, cost-effective)
• Hybrid: Support both with fallback logic

Step 3: Integrate with FastAPI Lifespan

Pattern:

from contextlib import asynccontextmanager

@asynccontextmanager
async def lifespan(app: FastAPI):
    # Startup
    memory_service = MemoryService()
    app.state.memory_service = memory_service
    logger.info("Memory service initialized")

    yield

    # Shutdown
    logger.info("Shutting down memory service")

Step 4: Add Memory Routes

Use template: templates/memory_routes.py

Register:

app.include_router(
    memory.router,
    prefix=f"{settings.API_V1_STR}/memory",
    tags=["memory"]
)

Step 5: Test Memory Service

# Run memory service tests
./scripts/test-memory.sh

Tests:

• Add conversation
• Search memories
• Get user summary
• Add preferences
• Error handling

Optimization Strategies

1. Memory Caching

Pattern:

from functools import lru_cache
from datetime import datetime, timedelta

class CachedMemoryService(MemoryService):
    def __init__(self):
        super().__init__()
        self._cache = {}
        self._cache_ttl = timedelta(minutes=5)

    async def search_memories(self, query: str, user_id: str, limit: int = 5):
        cache_key = f"{user_id}:{query}:{limit}"

        if cache_key in self._cache:
            cached_time, cached_result = self._cache[cache_key]
            if datetime.now() - cached_time < self._cache_ttl:
                return cached_result

        result = await super().search_memories(query, user_id, limit)
        self._cache[cache_key] = (datetime.now(), result)
        return result

2. Batch Operations

Pattern:

async def batch_add_conversations(
    conversations: List[Tuple[str, List[Dict]]],
    memory_service: MemoryService
):
    """Add multiple conversations efficiently"""
    tasks = [
        memory_service.add_conversation(user_id, messages)
        for user_id, messages in conversations
    ]
    return await asyncio.gather(*tasks)

3. Error Handling with Fallback

Pattern:

async def resilient_memory_add(
    user_id: str,
    messages: List[Dict],
    memory_service: MemoryService
):
    try:
        return await memory_service.add_conversation(user_id, messages)
    except Exception as e:
        logger.error(f"Memory add failed: {e}")
        # Fallback: Queue for retry or log to backup storage
        await backup_storage.save(user_id, messages)
        return None

Production Best Practices

1. Environment Configuration

Template: .env variables

# Hosted Mem0
MEM0_API_KEY=your_mem0_api_key

# Self-Hosted Configuration
QDRANT_HOST=localhost
QDRANT_PORT=6333
QDRANT_API_KEY=your_qdrant_key
OPENAI_API_KEY=your_openai_key

# Memory Settings
MEMORY_CACHE_TTL_SECONDS=300
MEMORY_SEARCH_LIMIT_DEFAULT=5
MEMORY_SEARCH_LIMIT_MAX=20

2. Monitoring and Logging

Pattern:

import logging
from datetime import datetime

logger = logging.getLogger(__name__)

class MonitoredMemoryService(MemoryService):
    async def add_conversation(self, user_id: str, messages: List[Dict], metadata=None):
        start_time = datetime.now()
        try:
            result = await super().add_conversation(user_id, messages, metadata)
            latency = (datetime.now() - start_time).total_seconds()
            logger.info(f"Memory add success: user={user_id}, latency={latency}s")
            return result
        except Exception as e:
            logger.error(f"Memory add failed: user={user_id}, error={e}")
            raise

3. Rate Limiting

Pattern:

from fastapi import HTTPException
from slowapi import Limiter
from slowapi.util import get_remote_address

limiter = Limiter(key_func=get_remote_address)

@router.post("/search")
@limiter.limit("10/minute")
async def search_memories(request: Request, ...):
    # Rate-limited memory search
    pass

Common Memory Patterns

1. Chat with Memory

Example: examples/chat_with_memory.py

Complete chat implementation with conversation history and context retrieval

Features:

• Automatic conversation storage
• Context-aware responses
• Session management
• Memory search integration

2. User Preference Management

Example: examples/user_preferences.py

User preference tracking and personalization

Features:

• Preference categorization
• Preference retrieval
• Behavioral learning
• Personalized AI responses

3. Multi-User Memory Isolation

Pattern:

async def get_user_memories_isolated(
    user_id: str,
    memory_service: MemoryService
):
    """Ensure memory isolation between users"""
    # All memory operations scoped to user_id
    memories = await memory_service.search_memories(
        query="...",
        user_id=user_id  # Always scoped
    )
    return memories

Vector Database Support

Supported Vector Databases

1. Qdrant (Recommended for Self-Hosted)

vector_store={
    "provider": "qdrant",
    "config": {
        "host": "localhost",
        "port": 6333,
        "api_key": settings.QDRANT_API_KEY
    }
}

2. Pinecone (Fully Managed)

vector_store={
    "provider": "pinecone",
    "config": {
        "api_key": settings.PINECONE_API_KEY,
        "environment": "us-west1-gcp"
    }
}

3. Chroma (Local Development)

vector_store={
    "provider": "chroma",
    "config": {
        "host": "localhost",
        "port": 8000
    }
}

Resources

Scripts:

• setup-mem0.sh - Install and configure Mem0
• test-memory.sh - Test memory service functionality

Templates:

• memory_service.py - Complete MemoryService implementation
• memory_client.py - Mem0 client configuration
• memory_middleware.py - Middleware and dependencies
• memory_routes.py - FastAPI routes for memory operations

Examples:

• chat_with_memory.py - Complete chat with memory
• user_preferences.py - User preference management
• memory_analytics.py - Memory usage analytics

---

Supported Configurations: Hosted Mem0, Self-Hosted (Qdrant, Pinecone, Chroma) FastAPI Version: 0.100+ Mem0 Version: 0.1.0+ Python Version: 3.9+

Best Practice: Start with hosted Mem0 for simplicity, migrate to self-hosted for cost optimization