MemU Python SDK

Official Python SDK for the MemU Cloud API - Manage structured, long-term memory for AI agents.

Features

• 🚀 Full Cloud API v3 Coverage - memorize, retrieve, categories, task status
• ⚡ Async/Sync Support - Both async and synchronous interfaces
• 🔄 Automatic Retry - Exponential backoff for failed requests
• ⏱️ Rate Limit Handling - Respects Retry-After headers
• 🛡️ Type Safety - Pydantic data models with full type hints
• 🎯 Custom Exceptions - Specific error types for different failure cases

Installation

Install from source:

git clone https://github.com/NevaMind-AI/memU-sdk-py.git
cd memU-sdk-py
pip install -e .

Quick Start

Get Your API Key

1. Sign up at memu.so
2. Navigate to your dashboard to obtain your API key

Basic Usage (Async)

import asyncio
from memu_sdk import MemUClient

async def main():
    # Initialize the client
    async with MemUClient(api_key="your_api_key") as client:
        # Memorize a conversation
        result = await client.memorize(
            conversation=[
                {"role": "user", "content": "I love Italian food, especially pasta."},
                {"role": "assistant", "content": "That's great! What's your favorite dish?"},
                {"role": "user", "content": "Carbonara is my absolute favorite!"}
            ],
            user_id="user_123",
            agent_id="my_assistant",
            wait_for_completion=True
        )

        print(f"Task ID: {result.task_id}")

        # Retrieve memories
        memories = await client.retrieve(
            query="What food does the user like?",
            user_id="user_123",
            agent_id="my_assistant"
        )

        print(f"Found {len(memories.items)} relevant memories")
        for item in memories.items:
            print(f"  - [{item.memory_type}] {item.content}")

asyncio.run(main())

Synchronous Usage

For scripts that don't use async/await:

from memu_sdk import MemUClient

# Initialize the client
client = MemUClient(api_key="your_api_key")

# Memorize a conversation (sync)
result = client.memorize_sync(
    conversation_text="User: I love pasta\nAssistant: Great choice!",
    user_id="user_123",
    agent_id="my_assistant"
)

# Retrieve memories (sync)
memories = client.retrieve_sync(
    query="What are the user's preferences?",
    user_id="user_123",
    agent_id="my_assistant"
)

# Clean up
client.close_sync()

API Reference

MemUClient

MemUClient(
    api_key: str,
    *,
    base_url: str = "https://api.memu.so",
    timeout: float = 60.0,
    max_retries: int = 3
)

Parameters:

• api_key: Your MemU API key (required)
• base_url: API base URL (default: https://api.memu.so)
• timeout: Request timeout in seconds (default: 60.0)
• max_retries: Maximum retry attempts for failed requests (default: 3)

Methods

memorize()

Memorize a conversation and extract structured memory.

async def memorize(
    *,
    conversation: list[dict] | None = None,
    conversation_text: str | None = None,
    user_id: str,  # Required
    agent_id: str,  # Required
    user_name: str = "User",
    agent_name: str = "Assistant",
    session_date: str | None = None,
    wait_for_completion: bool = False,
    poll_interval: float = 2.0,
    timeout: float | None = None,
) -> MemorizeResult

retrieve()

Retrieve relevant memories based on a query.

async def retrieve(
    query: str | list[dict],
    *,
    user_id: str,  # Required
    agent_id: str,  # Required
) -> RetrieveResult

list_categories()

List all memory categories for a user.

async def list_categories(
    *,
    user_id: str,  # Required
    agent_id: str | None = None,
) -> list[MemoryCategory]

get_task_status()

Get the status of an asynchronous memorization task.

async def get_task_status(task_id: str) -> TaskStatus

Synchronous Wrappers

All async methods have synchronous wrappers:

• memorize_sync() → wraps memorize()
• retrieve_sync() → wraps retrieve()
• list_categories_sync() → wraps list_categories()
• get_task_status_sync() → wraps get_task_status()
• close_sync() → wraps close()

Data Models

MemorizeResult

class MemorizeResult:
    task_id: str | None          # Task ID for async tracking
    resource: MemoryResource     # Created resource
    items: list[MemoryItem]      # Extracted memory items
    categories: list[MemoryCategory]  # Updated categories

RetrieveResult

class RetrieveResult:
    categories: list[MemoryCategory]  # Relevant categories
    items: list[MemoryItem]           # Relevant memory items
    resources: list[MemoryResource]   # Related raw resources
    next_step_query: str | None       # Rewritten query (if applicable)

MemoryItem

class MemoryItem:
    id: str | None               # Unique identifier
    summary: str | None          # Summary/description
    content: str | None          # Content text
    memory_type: str | None      # Type: profile, event, preference, etc.
    category_id: str | None      # Category ID
    category_name: str | None    # Category name
    score: float | None          # Relevance score (in retrieve)

MemoryCategory

class MemoryCategory:
    id: str | None               # Unique identifier
    name: str | None             # Category name (e.g., 'personal info')
    summary: str | None          # Summary of content
    content: str | None          # Full content
    description: str | None      # Description
    item_count: int | None       # Number of items
    score: float | None          # Relevance score (in retrieve)

TaskStatus

class TaskStatus:
    task_id: str                 # Task identifier
    status: TaskStatusEnum       # PENDING, PROCESSING, COMPLETED, SUCCESS, FAILED
    progress: float | None       # Progress percentage (0-100)
    message: str | None          # Status message or error
    result: dict | None          # Task result when completed

Error Handling

The SDK provides specific exception types for different error cases:

from memu_sdk.client import (
    MemUClientError,        # Base exception
    MemUAuthenticationError, # Invalid API key (401)
    MemURateLimitError,      # Rate limit exceeded (429)
    MemUNotFoundError,       # Resource not found (404)
    MemUValidationError,     # Request validation failed (422)
)

try:
    result = await client.memorize(
        conversation_text="Hello",
        user_id="user_123",
        agent_id="agent_456"
    )
except MemUAuthenticationError:
    print("Invalid API key")
except MemURateLimitError as e:
    print(f"Rate limited. Retry after {e.retry_after} seconds")
except MemUValidationError as e:
    print(f"Validation error: {e.response}")
except MemUClientError as e:
    print(f"API error: {e.message}")

Examples

See the examples directory for complete working examples:

• demo.py - Complete workflow demonstration

Development

Setup

# Clone the repository
git clone https://github.com/NevaMind-AI/memU-sdk-py.git
cd memU-sdk-py

# Install with development dependencies
pip install -e ".[dev]"

Running Tests

# Run unit tests
pytest tests/test_sdk.py

# Run integration tests (requires API key)
MEMU_API_KEY=your_key python tests/test_integration.py

Code Quality

# Format code
ruff format .

# Lint
ruff check .

# Type check
mypy src/memu_sdk

Support

• 📚 Full API Documentation
• 💬 Discord Community
• 🐛 Report Issues

License

MIT License - see LICENSE for details.