memoryrelay 0.3.0

Official Python SDK for MemoryRelay - persistent memory for AI agents with v2 async API support

MemoryRelay Python SDK

Official Python client for MemoryRelay - persistent memory for AI agents.

Features

• 🚀 Simple API - Intuitive Pythonic interface
• ⚡ v2 Async API - 60-600x faster with background processing
• ⚡ Async/Await Support - Full async client for high-performance applications
• 🔍 Semantic Search - Vector-based memory retrieval
• 📦 Batch Operations - Create multiple memories efficiently
• 🏷️ Entity Tracking - Automatic relationship management
• 🔐 Type Safe - Full Pydantic models and type hints
• 🐍 Python 3.9+ - Modern Python support
• ✅ Production Tested - Verified against live API

Installation

pip install memoryrelay

Quick Start

Sync Client

from memoryrelay import MemoryRelay

# Initialize client
client = MemoryRelay(api_key="mem_your_api_key_here")

# Create a memory
memory = client.memories.create(
    content="User prefers dark mode",
    agent_id="my-agent"
)

# Search memories
results = client.memories.search(
    query="user preferences",
    limit=5
)

for result in results:
    print(f"Score: {result.score:.3f}")
    print(f"Content: {result.memory.content}")

Async Client

import asyncio
from memoryrelay import AsyncMemoryRelay

async def main():
    async with AsyncMemoryRelay(api_key="mem_your_api_key_here") as client:
        # Create a memory
        memory = await client.memories.create(
            content="User prefers dark mode",
            agent_id="my-agent"
        )
        
        # Search memories
        results = await client.memories.search(
            query="user preferences",
            limit=5
        )
        
        for result in results:
            print(f"Score: {result.score:.3f}")
            print(f"Content: {result.memory.content}")

asyncio.run(main())

v2 Async API (60-600x Faster)

The v2 API returns immediately (202 Accepted) while embedding generation happens in the background.

from memoryrelay import MemoryRelay

client = MemoryRelay(api_key="mem_your_api_key_here")

# Strategy 1: Fire-and-forget (fastest, no waiting)
response = client.memories.create_async(
    content="User prefers dark mode",
    agent_id="my-agent"
)
print(f"Memory {response.id} queued (job: {response.job_id})")
# API response in <50ms! Memory will be ready in ~3s

# Strategy 2: Poll until ready (drop-in v1 replacement)
response = client.memories.create_async(
    content="User's favorite color is blue",
    agent_id="my-agent"
)
memory = client.memories.wait_for_ready(response.id, timeout=10)
# Blocks until embedding is generated (still faster than v1!)

# Strategy 3: Create and wait (convenience helper)
memory = client.memories.create_and_wait(
    content="User timezone is America/New_York",
    agent_id="my-agent",
    timeout=10
)
# Same interface as v1, but uses v2 internally (faster!)

Performance:

• v1 (sync): 2-30s per memory (blocking)
• v2 (async): <50ms API response + 2-5s background processing
• Speedup: 60-600x faster API response time

See examples/v2_async_api.py for detailed examples.

Usage

Initialize Client

Sync Client

from memoryrelay import MemoryRelay

# Basic initialization
client = MemoryRelay(api_key="mem_your_api_key_here")

# With custom configuration
client = MemoryRelay(
    api_key="mem_your_api_key_here",
    base_url="https://api.memoryrelay.net",  # Optional
    timeout=30.0,  # Request timeout in seconds
    max_retries=3  # Max retries for failed requests
)

# Or use context manager (recommended)
with MemoryRelay(api_key="mem_...") as client:
    # Your code here
    pass

Async Client

from memoryrelay import AsyncMemoryRelay

# Basic initialization
client = AsyncMemoryRelay(api_key="mem_your_api_key_here")

# Use async context manager (recommended)
async with AsyncMemoryRelay(api_key="mem_...") as client:
    # Your async code here
    memory = await client.memories.create(...)

Create Memories

Sync

# Create a single memory
memory = client.memories.create(
    content="User completed Python tutorial",
    agent_id="learning-agent",
    metadata={"course": "python-101", "completed": True}
)

# Batch create (faster for multiple memories)
response = client.memories.create_batch([
    {"content": "Memory 1", "agent_id": "agent-1"},
    {"content": "Memory 2", "agent_id": "agent-1"},
    {"content": "Memory 3", "agent_id": "agent-1"}
])

print(f"Created {response.succeeded}/{response.total} memories")
print(f"Took {response.timing['total_ms']:.0f}ms")

Async

# Create a single memory
memory = await client.memories.create(
    content="User completed Python tutorial",
    agent_id="learning-agent",
    metadata={"course": "python-101", "completed": True}
)

# Batch create (faster for multiple memories)
response = await client.memories.create_batch([
    {"content": "Memory 1", "agent_id": "agent-1"},
    {"content": "Memory 2", "agent_id": "agent-1"},
    {"content": "Memory 3", "agent_id": "agent-1"}
])

print(f"Created {response.succeeded}/{response.total} memories")
print(f"Took {response.timing['total_ms']:.0f}ms")

Search Memories

# Semantic search
results = client.memories.search(
    query="what programming languages does the user know?",
    agent_id="my-agent",
    limit=10,
    min_score=0.7  # Only return results with score >= 0.7
)

# With metadata filtering
results = client.memories.search(
    query="completed courses",
    metadata_filter={"completed": True}
)

Update & Delete

# Update memory
updated = client.memories.update(
    memory_id="mem_abc123",
    content="Updated content",
    metadata={"updated": True}
)

# Delete memory
client.memories.delete(memory_id="mem_abc123")

List Memories

# List all memories for an agent
memories = client.memories.list(
    agent_id="my-agent",
    limit=100,
    offset=0
)

# List with user filter
memories = client.memories.list(
    user_id="user_123",
    limit=50
)

Entity Management

# Create entity
entity = client.entities.create(
    entity_type="person",
    name="John Doe"
)

# Link entity to memory
client.entities.link(
    entity_id=entity.id,
    memory_id=memory.id
)

# List entities
entities = client.entities.list(
    agent_id="my-agent",
    entity_type="person"
)

Health Check

health = client.health()
print(f"API Status: {health.status}")
print(f"Version: {health.version}")
print(f"Services: {health.services}")

Error Handling

from memoryrelay import (
    MemoryRelay,
    AuthenticationError,
    RateLimitError,
    NotFoundError,
    ValidationError,
    APIError,
)

try:
    memory = client.memories.create(
        content="Test memory",
        agent_id="my-agent"
    )
except AuthenticationError:
    print("Invalid API key")
except RateLimitError as e:
    print(f"Rate limit exceeded. Retry after {e.retry_after}s")
except NotFoundError:
    print("Resource not found")
except ValidationError as e:
    print(f"Invalid request: {e.message}")
except APIError as e:
    print(f"API error: {e.message} (status: {e.status_code})")

Examples

See the examples/ directory for more usage examples:

• basic_usage.py - Sync client CRUD operations
• context_manager.py - Using context managers
• async_usage.py - Async/await operations

Development

Setup

git clone https://github.com/memoryrelay/python-sdk.git
cd python-sdk
python -m venv venv
source venv/bin/activate  # or `venv\Scripts\activate` on Windows
pip install -e ".[dev]"

Testing

pytest
pytest --cov=memoryrelay  # With coverage

Code Quality

black .
ruff check .
mypy memoryrelay

API Reference

Full API documentation available at docs.memoryrelay.ai

Support

• Documentation: docs.memoryrelay.ai
• GitHub Issues: github.com/memoryrelay/python-sdk/issues
• Email: [e-mail]

License

MIT License - see LICENSE for details.