Python Quick Start

This guide walks through the core operations of the MemoryLayer Python SDK.

Prerequisites

• Python 3.12+
• MemoryLayer server running (memorylayer serve)

Install the SDK

pip install memorylayer-client

Connect to the Server

Async Client

from memorylayer import MemoryLayerClient

async with MemoryLayerClient(
    base_url="http://localhost:61001",
    api_key="your-api-key",
    workspace_id="my-workspace"
) as client:
    # Use the client...
    pass

Sync Client

from memorylayer import SyncMemoryLayerClient

with SyncMemoryLayerClient(
    base_url="http://localhost:61001",
    api_key="your-api-key",
    workspace_id="my-workspace"
) as client:
    # Use the client (no await needed)
    memory = client.remember(
        content="User prefers Python",
        type=MemoryType.SEMANTIC
    )

Or use the convenience function:

from memorylayer import sync_client

with sync_client(
    base_url="http://localhost:61001",
    api_key="your-api-key",
    workspace_id="my-workspace"
) as client:
    results = client.recall("coding preferences")

Note

All examples below use the async client. The sync client has identical methods without await.

Remember (Store Memories)

from memorylayer import MemoryType, MemorySubtype

# Basic storage
memory = await client.remember(
    content="User prefers FastAPI over Flask",
    type=MemoryType.SEMANTIC,
)

# With full options
memory = await client.remember(
    content="User prefers FastAPI over Flask",
    type=MemoryType.SEMANTIC,
    subtype=MemorySubtype.PREFERENCE,
    importance=0.8,
    tags=["preferences", "frameworks"],
    metadata={"source": "conversation"},
    context_id="project-alpha",
    user_id="user_42",
)

Recall (Search Memories)

from memorylayer import RecallMode, SearchTolerance

# Simple search
results = await client.recall(
    query="what frameworks does the user prefer?",
    limit=5
)

# Advanced search with filters
results = await client.recall(
    query="what frameworks does the user prefer?",
    types=[MemoryType.SEMANTIC],
    subtypes=[MemorySubtype.PREFERENCE],
    tags=["frameworks"],
    mode=RecallMode.RAG,
    limit=10,
    min_relevance=0.7,
    recency_weight=0.3,
    tolerance=SearchTolerance.MODERATE,
    include_associations=True,
    traverse_depth=2,
    max_expansion=20,
    created_after="2025-01-01T00:00:00Z",
)

for memory in results.memories:
    print(f"{memory.content} (relevance: {memory.relevance_score})")

Reflect (Synthesize Insights)

reflection = await client.reflect(
    query="summarize everything about the user's development workflow",
    max_tokens=500,
    include_sources=True
)

print(reflection.reflection)
print(f"Confidence: {reflection.confidence}")
print(f"Based on {len(reflection.source_memories)} memories")

Memory Management

# Get a specific memory
memory = await client.get_memory("mem_123")

# Update a memory
updated = await client.update_memory(
    "mem_123",
    content="Updated content",
    importance=0.9,
    tags=["preferences", "high-priority"],
    metadata={"reviewed": True}
)

# Soft delete (archive)
await client.forget("mem_123")

# Hard delete (permanent)
await client.forget("mem_123", hard=True)

# Decay a memory's importance
decayed = await client.decay("mem_123", decay_rate=0.1)

# Trace memory provenance
trace = await client.trace_memory("mem_123")
print(trace["chain"])

Associate (Link Memories)

from memorylayer import RelationshipType

# Create an association
association = await client.associate(
    source_id="mem_problem_123",
    target_id="mem_solution_456",
    relationship=RelationshipType.SOLVES,
    strength=0.9,
    metadata={"context": "debugging session"}
)

# Get associations for a memory
associations = await client.get_associations(
    "mem_123",
    direction="both"  # "outgoing", "incoming", or "both"
)

for assoc in associations:
    print(f"{assoc.source_id} --{assoc.relationship}--> {assoc.target_id}")

Sessions (Working Memory)

Sessions provide temporary working memory that persists across API calls and can be committed to long-term storage.

# Create a session (auto-sets as active session)
session = await client.create_session(
    ttl_seconds=3600,
    workspace_id="my-workspace",
    context_id="project-alpha",
)

# Store working memory
await client.set_context(
    session.id,
    "current_task",
    {"description": "Debugging auth", "file": "auth.py"}
)

# Retrieve working memory
context = await client.get_context(session.id, ["current_task"])

# Extend session TTL
await client.touch_session(session.id)

# Commit to long-term storage
result = await client.commit_session(
    session.id,
    min_importance=0.5,
    deduplicate=True,
    max_memories=50,
    categories=["decisions", "preferences"]
)
print(f"Created {result['memories_created']} memories")

# List sessions
sessions = await client.list_sessions(
    workspace_id="my-workspace",
    include_expired=False
)

# Delete session
await client.delete_session(session.id)

Session Management Helpers

# Manually set/get/clear the active session
client.set_session("sess_abc123")
current = client.get_session_id()  # "sess_abc123"
client.clear_session()

Session Briefing

Get a summary of recent activity:

briefing = await client.get_briefing(
    lookback_minutes=120,
    detail_level="full",
    limit=20,
    include_memories=True,
    include_contradictions=True,
)
print(briefing.workspace_summary)
for activity in briefing.recent_activity:
    print(f"{activity.timestamp}: {activity.summary}")

Workspace Management

# Create workspace
workspace = await client.create_workspace("my-project")

# Get workspace details
workspace = await client.get_workspace("ws_123")

# Update workspace
workspace = await client.update_workspace(
    "ws_123",
    name="New Name",
    settings={"key": "value"}
)

# Get workspace schema
schema = await client.get_workspace_schema("ws_123")
print(schema["relationship_types"])
print(schema["memory_subtypes"])

Contexts

Contexts provide logical grouping within a workspace:

# Create a context
context = await client.create_context(
    "ws_123",
    name="project-alpha",
    description="Memories for Project Alpha"
)

# List contexts
contexts = await client.list_contexts("ws_123")

Batch Operations

results = await client.batch_memories([
    {"type": "create", "data": {"content": "Memory 1", "importance": 0.7}},
    {"type": "create", "data": {"content": "Memory 2", "importance": 0.8}},
    {"type": "update", "data": {"memory_id": "mem_123", "content": "Updated"}},
    {"type": "delete", "data": {"memory_id": "mem_old", "hard": False}}
])
print(f"Successful: {results['successful']}, Failed: {results['failed']}")

Export and Import

# Export workspace data
data = await client.export_workspace("ws_123", include_associations=True)
print(f"Exported {data['total_memories']} memories")

# Stream export for large workspaces
async for line in client.export_workspace_stream("ws_123"):
    if line["type"] == "memory":
        print(f"Memory: {line['data']['content']}")

# Import into another workspace
result = await client.import_workspace("ws_target", data)
print(f"Imported {result['imported']} memories")

Chat Threads

Chat threads store conversation history and can decompose messages into memories:

# Create a thread
thread = await client.create_thread(
    user_id="user_123",
    title="Debugging Session",
    context_id="project-alpha"
)

# Append messages
messages = await client.append_messages(
    thread.id,
    [
        {"role": "user", "content": "How do I fix the auth bug?"},
        {"role": "assistant", "content": "Check the token validation logic."}
    ]
)

# Retrieve messages
messages = await client.get_messages(thread.id, limit=50)

# Get thread with all messages
full = await client.get_thread_full(thread.id)

# Decompose messages into memories
result = await client.decompose_thread(thread.id)
print(f"Created {result.memories_created} memories from {result.messages_processed} messages")

# List and delete threads
threads = await client.list_threads(limit=20)
await client.delete_thread(thread.id)

Context Environment

The context environment provides a server-side Python sandbox for executing code, querying LLMs, and running autonomous reasoning loops over memories.

Setup

Context environments are session-scoped. Create a session first:

session = await client.create_session(workspace_id="my-workspace")

Load and Analyze Memories

# Load memories into the sandbox
await client.context_load(
    var="project_memories",
    query="project architecture decisions",
    limit=50
)

# Run code against loaded memories
result = await client.context_exec("""
decisions = [m for m in project_memories if m.get('subtype') == 'decision']
summary = f"Found {len(decisions)} architecture decisions"
""", result_var="summary")

print(result["result"])

Query LLM with Sandbox Context

# Ask the server-side LLM to analyze sandbox data
answer = await client.context_query(
    prompt="What are the key architecture decisions and their rationale?",
    variables=["decisions"]
)
print(answer["response"])

Inspect Sandbox State

# View all variables
state = await client.context_inspect()
print(state["variables"])

# Inspect a specific variable
detail = await client.context_inspect(variable="decisions", preview_chars=500)
print(detail["preview"])

Autonomous Reasoning (RLM)

Run a Recursive Language Model loop that iteratively reasons over memories:

result = await client.context_rlm(
    goal="Identify recurring error patterns and recommend fixes",
    memory_query="errors bugs fixes",
    max_iterations=10,
    detail_level="standard"
)
print(result["result"])
print(f"Completed in {result['iterations']} iterations")

Cleanup

await client.context_cleanup()

Error Handling

from memorylayer import (
    MemoryLayerError,
    AuthenticationError,
    AuthorizationError,
    NotFoundError,
    ValidationError,
    RateLimitError,
    ServerError,
    EnterpriseRequiredError,
)

try:
    memory = await client.get_memory("mem_123")
except NotFoundError:
    print("Memory not found")
except AuthenticationError:
    print("Invalid API key")
except AuthorizationError:
    print("Access denied")
except RateLimitError:
    print("Rate limit exceeded")
except EnterpriseRequiredError as e:
    print(f"Enterprise feature required: {e.feature}")
except ServerError as e:
    print(f"Server error: {e.status_code}")
except MemoryLayerError as e:
    print(f"Error {e.status_code}: {e.message}")