Python SDK Examples#

from memofai import create_moa_client
import os

client = create_moa_client(api_token=os.getenv('MEMOFAI_TOKEN'))

# Setup
workspace = client.workspaces.create({'name': 'My App'})
bot = client.bots.create({
    'name': 'Assistant',
    'workspace_id': workspace['id']
})

# Store
client.memories.create({
    'bot_id': bot['id'],
    'content': 'User loves Python'
})

# Search
results = client.memories.search({
    'bot_id': bot['id'],
    'query': 'programming language'
})

print(results[0]['content'])

class PreferencesManager:
    """Manage user preferences"""
    
    def __init__(self, user_id: str, bot_id: str):
        self.user_id = user_id
        self.bot_id = bot_id
        self.client = create_moa_client(api_token=os.getenv('MEMOFAI_TOKEN'))
    
    def set_preference(self, key: str, value: str):
        """Set a preference"""
        return self.client.memories.create({
            'bot_id': self.bot_id,
            'content': f"User {self.user_id} preference: {key} = {value}"
        })
    
    def get_preferences(self):
        """Get all preferences"""
        return self.client.memories.search({
            'bot_id': self.bot_id,
            'query': f'user {self.user_id} preferences',
            'limit': 20
        })

# Usage
prefs = PreferencesManager('user_123', 'bot_xyz')
prefs.set_preference('theme', 'dark')
prefs.set_preference('language', 'Python')
all_prefs = prefs.get_preferences()

from fastapi import FastAPI, HTTPException, Depends
from pydantic import BaseModel
from memofai import create_moa_client
from typing import List, Optional
import os

app = FastAPI(title="Memory API")

# Initialize client
moa_client = create_moa_client(
    api_token=os.getenv('MEMOFAI_TOKEN'),
    environment='production'
)

# Models
class BotCreate(BaseModel):
    name: str
    workspace_id: str
    description: Optional[str] = None

class MemoryCreate(BaseModel):
    bot_id: str
    content: str

class SearchRequest(BaseModel):
    bot_id: str
    query: str
    limit: int = 5

# Endpoints
@app.post("/bots")
def create_bot(bot: BotCreate):
    """Create a new bot"""
    try:
        result = moa_client.bots.create({
            'name': bot.name,
            'workspace_id': bot.workspace_id,
            'description': bot.description
        })
        return result
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))

@app.get("/bots")
def list_bots():
    """List all bots"""
    try:
        return moa_client.bots.list()
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))

@app.post("/memories")
def create_memory(memory: MemoryCreate):
    """Store a memory"""
    try:
        result = moa_client.memories.create({
            'bot_id': memory.bot_id,
            'content': memory.content
        })
        return result
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))

@app.post("/search")
def search_memories(request: SearchRequest):
    """Search memories"""
    try:
        results = moa_client.memories.search({
            'bot_id': request.bot_id,
            'query': request.query,
            'limit': request.limit
        })
        return {'results': results, 'count': len(results)}
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))

if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8000)

# settings.py
from memofai import create_moa_client
import os

MEMOFAI_CLIENT = create_moa_client(
    api_token=os.getenv('MEMOFAI_TOKEN'),
    environment='production'
)

# models.py
from django.db import models

class UserProfile(models.Model):
    user = models.OneToOneField('auth.User', on_delete=models.CASCADE)
    memofai_bot_id = models.CharField(max_length=100, blank=True)
    
    def get_or_create_bot(self):
        if not self.memofai_bot_id:
            from django.conf import settings
            
            bot = settings.MEMOFAI_CLIENT.bots.create({
                'name': f'user_{self.user.id}',
                'workspace_id': settings.MEMOFAI_WORKSPACE_ID
            })
            self.memofai_bot_id = bot['id']
            self.save()
        
        return self.memofai_bot_id

# views.py
from django.http import JsonResponse
from django.conf import settings
from django.views.decorators.http import require_http_methods
import json

@require_http_methods(["POST"])
def store_memory(request):
    user_profile = request.user.userprofile
    bot_id = user_profile.get_or_create_bot()
    
    data = json.loads(request.body)
    
    memory = settings.MEMOFAI_CLIENT.memories.create({
        'bot_id': bot_id,
        'content': data['content']
    })
    
    return JsonResponse(memory)

@require_http_methods(["POST"])
def search_memories(request):
    user_profile = request.user.userprofile
    bot_id = user_profile.get_or_create_bot()
    
    data = json.loads(request.body)
    
    results = settings.MEMOFAI_CLIENT.memories.search({
        'bot_id': bot_id,
        'query': data['query'],
        'limit': data.get('limit', 5)
    })
    
    return JsonResponse({'results': results})

from flask import Flask, request, jsonify, session
from memofai import create_moa_client
from functools import wraps
import os

app = Flask(__name__)
app.secret_key = os.getenv('SECRET_KEY')

# Initialize client
moa_client = create_moa_client(
    api_token=os.getenv('MEMOFAI_TOKEN'),
    environment='production'
)

WORKSPACE_ID = os.getenv('MEMOFAI_WORKSPACE_ID')

# Helper to get user's bot
def get_user_bot():
    user_id = session.get('user_id')
    if not user_id:
        return None
    
    bots = moa_client.bots.list()
    user_bot = next(
        (b for b in bots if b['name'] == f'user_{user_id}'),
        None
    )
    
    if not user_bot:
        user_bot = moa_client.bots.create({
            'name': f'user_{user_id}',
            'workspace_id': WORKSPACE_ID
        })
    
    return user_bot

# Require auth decorator
def require_auth(f):
    @wraps(f)
    def decorated(*args, **kwargs):
        if 'user_id' not in session:
            return jsonify({'error': 'Unauthorized'}), 401
        return f(*args, **kwargs)
    return decorated

@app.route('/login', methods=['POST'])
def login():
    data = request.json
    session['user_id'] = data['user_id']
    return jsonify({'status': 'logged in'})

@app.route('/memories', methods=['POST'])
@require_auth
def create_memory():
    bot = get_user_bot()
    data = request.json
    
    memory = moa_client.memories.create({
        'bot_id': bot['id'],
        'content': data['content']
    })
    
    return jsonify(memory)

@app.route('/search', methods=['POST'])
@require_auth
def search():
    bot = get_user_bot()
    data = request.json
    
    results = moa_client.memories.search({
        'bot_id': bot['id'],
        'query': data['query'],
        'limit': data.get('limit', 5)
    })
    
    return jsonify({'results': results})

@app.route('/memories', methods=['GET'])
@require_auth
def list_memories():
    bot = get_user_bot()
    
    memories = moa_client.memories.list({
        'bot_id': bot['id']
    })
    
    return jsonify({'memories': memories})

if __name__ == '__main__':
    app.run(debug=True)

from functools import lru_cache
from datetime import datetime, timedelta

class CachedMemoryManager:
    """Memory manager with caching"""
    
    def __init__(self, bot_id: str, cache_ttl: int = 300):
        self.bot_id = bot_id
        self.cache_ttl = cache_ttl
        self.client = create_moa_client(api_token=os.getenv('MEMOFAI_TOKEN'))
        self._cache = {}
        self._cache_times = {}
    
    def _is_cache_valid(self, key: str) -> bool:
        if key not in self._cache_times:
            return False
        age = (datetime.now() - self._cache_times[key]).total_seconds()
        return age < self.cache_ttl
    
    def search(self, query: str, limit: int = 5):
        """Search with caching"""
        cache_key = f"{query}:{limit}"
        
        if self._is_cache_valid(cache_key):
            return self._cache[cache_key]
        
        results = self.client.memories.search({
            'bot_id': self.bot_id,
            'query': query,
            'limit': limit
        })
        
        self._cache[cache_key] = results
        self._cache_times[cache_key] = datetime.now()
        
        return results
    
    def create(self, content: str):
        """Create and invalidate cache"""
        result = self.client.memories.create({
            'bot_id': self.bot_id,
            'content': content
        })
        
        # Invalidate cache
        self._cache.clear()
        self._cache_times.clear()
        
        return result

from concurrent.futures import ThreadPoolExecutor, as_completed

class BatchMemoryManager:
    """Batch memory operations"""
    
    def __init__(self, bot_id: str, max_workers: int = 5):
        self.bot_id = bot_id
        self.max_workers = max_workers
        self.client = create_moa_client(api_token=os.getenv('MEMOFAI_TOKEN'))
    
    def batch_create(self, contents: list[str]):
        """Create multiple memories"""
        def create_one(content):
            return self.client.memories.create({
                'bot_id': self.bot_id,
                'content': content
            })
        
        with ThreadPoolExecutor(max_workers=self.max_workers) as executor:
            futures = [executor.submit(create_one, c) for c in contents]
            results = []
            
            for future in as_completed(futures):
                try:
                    results.append(future.result())
                except Exception as e:
                    print(f"Error: {e}")
            
            return results
    
    def batch_search(self, queries: list[str]):
        """Search multiple queries"""
        def search_one(query):
            return self.client.memories.search({
                'bot_id': self.bot_id,
                'query': query,
                'limit': 5
            })
        
        with ThreadPoolExecutor(max_workers=self.max_workers) as executor:
            futures = [executor.submit(search_one, q) for q in queries]
            results = {}
            
            for query, future in zip(queries, futures):
                try:
                    results[query] = future.result()
                except Exception as e:
                    results[query] = []
                    print(f"Error searching '{query}': {e}")
            
            return results

#!/usr/bin/env python3
import click
from memofai import create_moa_client
import os

client = create_moa_client(api_token=os.getenv('MEMOFAI_TOKEN'))

@click.group()
def cli():
    """Memof.ai CLI Tool"""
    pass

@cli.command()
@click.argument('bot_id')
@click.argument('content')
def store(bot_id, content):
    """Store a memory"""
    memory = client.memories.create({
        'bot_id': bot_id,
        'content': content
    })
    click.echo(f"Stored: {memory['id']}")

@cli.command()
@click.argument('bot_id')
@click.argument('query')
@click.option('--limit', default=5, help='Number of results')
def search(bot_id, query, limit):
    """Search memories"""
    results = client.memories.search({
        'bot_id': bot_id,
        'query': query,
        'limit': limit
    })
    
    click.echo(f"Found {len(results)} results:")
    for i, result in enumerate(results, 1):
        click.echo(f"{i}. {result['content']}")

@cli.command()
def list_bots():
    """List all bots"""
    bots = client.bots.list()
    for bot in bots:
        click.echo(f"{bot['name']} (ID: {bot['id']})")

if __name__ == '__main__':
    cli()