Python API Reference

MemoryLayerClient

The main async client class for interacting with MemoryLayer. Use as an async context manager.

Constructor

MemoryLayerClient(
    base_url: str = "http://localhost:61001",
    api_key: str | None = None,
    workspace_id: str | None = None,
    session_id: str | None = None,
    timeout: float = 30.0,
)

Parameters:

Parameter	Type	Default	Description
base_url	str	"http://localhost:61001"	Server URL
api_key	str | None	None	API key for authentication
workspace_id	str | None	None	Default workspace ID
session_id	str | None	None	Active session ID (sent as X-Session-ID header)
timeout	float	30.0	Request timeout in seconds

Usage:

async with MemoryLayerClient(
    base_url="http://localhost:61001",
    api_key="your-key",
    workspace_id="ws_123"
) as client:
    memory = await client.remember("Hello world")

---

Memory Operations

remember()

Store a new memory.

async def remember(
    content: str,
    type: MemoryType | str | None = None,
    subtype: MemorySubtype | str | None = None,
    importance: float = 0.5,
    tags: list[str] | None = None,
    metadata: dict | None = None,
    context_id: str | None = None,
    user_id: str | None = None,
) -> Memory

Parameters:

Parameter	Type	Default	Description
content	str	required	Memory content to store
type	MemoryType | str | None	None	Cognitive memory type
subtype	MemorySubtype | str | None	None	Domain subtype
importance	float	0.5	Importance score 0.0-1.0
tags	list[str] | None	None	Tags for categorization
metadata	dict | None	None	Additional metadata
context_id	str | None	None	Context within workspace
user_id	str | None	None	User ID to associate

Returns: Memory

memory = await client.remember(
    content="User prefers concise code comments",
    type=MemoryType.SEMANTIC,
    subtype=MemorySubtype.PREFERENCE,
    importance=0.8,
    tags=["preferences", "coding-style"]
)

recall()

Search memories by semantic query.

async def recall(
    query: str,
    types: list[MemoryType | str] | None = None,
    subtypes: list[MemorySubtype | str] | None = None,
    tags: list[str] | None = None,
    mode: RecallMode | str | None = None,
    limit: int = 10,
    min_relevance: float | None = None,
    recency_weight: float | None = None,
    tolerance: SearchTolerance | str | None = None,
    include_associations: bool | None = None,
    traverse_depth: int | None = None,
    max_expansion: int | None = None,
    created_after: str | None = None,
    created_before: str | None = None,
    user_id: str | None = None,
) -> RecallResult

Parameters:

Parameter	Type	Default	Description
query	str	required	Natural language search query
types	list[MemoryType] | None	None	Filter by memory types
subtypes	list[MemorySubtype] | None	None	Filter by subtypes
tags	list[str] | None	None	Filter by tags
mode	RecallMode | None	None	Retrieval mode (rag, llm, hybrid)
limit	int	10	Maximum memories to return
min_relevance	float | None	None	Minimum relevance score 0.0-1.0
recency_weight	float | None	None	Weight for recency in ranking 0.0-1.0
tolerance	SearchTolerance | None	None	Search precision (loose, moderate, strict)
include_associations	bool | None	None	Include linked memories
traverse_depth	int | None	None	Multi-hop graph traversal depth
max_expansion	int | None	None	Max memories via graph expansion
created_after	str | None	None	ISO 8601 datetime filter
created_before	str | None	None	ISO 8601 datetime filter
user_id	str | None	None	Filter by user ID

Returns: RecallResult with fields: memories: list[Memory], total_count: int, query_tokens: int | None, search_latency_ms: int | None

results = await client.recall(
    query="coding preferences",
    types=[MemoryType.SEMANTIC],
    limit=5,
    min_relevance=0.7
)

reflect()

Synthesize insights across memories.

async def reflect(
    query: str,
    max_tokens: int = 500,
    include_sources: bool = True,
) -> ReflectResult

Parameters:

Parameter	Type	Default	Description
query	str	required	What to reflect on
max_tokens	int	500	Maximum tokens in reflection
include_sources	bool	True	Include source memory IDs

Returns: ReflectResult with fields: reflection: str, source_memories: list[str], confidence: float, tokens_processed: int | None

reflection = await client.reflect(
    query="summarize user's development workflow",
    max_tokens=300
)

get_memory()

Get a specific memory by ID.

async def get_memory(memory_id: str) -> Memory

memory = await client.get_memory("mem_123")

update_memory()

Update a memory’s properties.

async def update_memory(
    memory_id: str,
    content: str | None = None,
    importance: float | None = None,
    tags: list[str] | None = None,
    metadata: dict | None = None,
) -> Memory

Parameters:

Parameter	Type	Default	Description
memory_id	str	required	Memory ID
content	str | None	None	New content
importance	float | None	None	New importance score
tags	list[str] | None	None	New tags
metadata	dict | None	None	New metadata

Returns: Memory

updated = await client.update_memory("mem_123", importance=0.9)

forget()

Delete or archive a memory.

async def forget(memory_id: str, hard: bool = False) -> bool

Parameter	Type	Default	Description
memory_id	str	required	Memory ID
hard	bool	False	True for permanent delete, False for soft delete

Returns: True if successful

decay()

Reduce a memory’s importance.

async def decay(memory_id: str, decay_rate: float = 0.1) -> Memory

Parameter	Type	Default	Description
memory_id	str	required	Memory ID
decay_rate	float	0.1	Decay rate 0.0-1.0

Returns: Memory with updated importance

trace_memory()

Get memory provenance and association chain.

async def trace_memory(memory_id: str) -> dict

Returns: Dict with provenance information including source resource, category membership, and association chain.

batch_memories()

Perform multiple operations in one request.

async def batch_memories(operations: list[dict]) -> dict

Supported operation types:

• {"type": "create", "data": {"content": "...", "importance": 0.5}}
• {"type": "update", "data": {"memory_id": "...", "content": "..."}}
• {"type": "delete", "data": {"memory_id": "...", "hard": False}}

Returns: Dict with successful, failed, and per-operation results.

results = await client.batch_memories([
    {"type": "create", "data": {"content": "Memory 1"}},
    {"type": "delete", "data": {"memory_id": "mem_old"}}
])

---

Association Operations

associate()

Create a relationship between memories.

async def associate(
    source_id: str,
    target_id: str,
    relationship: RelationshipType | str,
    strength: float = 0.5,
    metadata: dict | None = None,
) -> Association

Parameters:

Parameter	Type	Default	Description
source_id	str	required	Source memory ID
target_id	str	required	Target memory ID
relationship	RelationshipType | str	required	Relationship type
strength	float	0.5	Strength 0.0-1.0
metadata	dict | None	None	Additional metadata

Returns: Association

assoc = await client.associate(
    "mem_123", "mem_456",
    RelationshipType.SOLVES, strength=0.9
)

get_associations()

Get associations for a memory.

async def get_associations(
    memory_id: str,
    direction: str = "both",
) -> list[Association]

Parameter	Type	Default	Description
memory_id	str	required	Memory ID
direction	str	"both"	"outgoing", "incoming", or "both"

Returns: list[Association]

---

Session Operations

create_session()

Create a new working memory session.

async def create_session(
    ttl_seconds: int = 3600,
    workspace_id: str | None = None,
    context_id: str | None = None,
    auto_set_session: bool = True,
) -> Session

Parameters:

Parameter	Type	Default	Description
ttl_seconds	int	3600	Time to live in seconds
workspace_id	str | None	None	Workspace ID (auto-created if needed)
context_id	str | None	None	Context ID (defaults to _default)
auto_set_session	bool	True	Auto-set as active session

Returns: Session

get_session()

async def get_session(session_id: str) -> Session

list_sessions()

async def list_sessions(
    workspace_id: str | None = None,
    context_id: str | None = None,
    include_expired: bool = False,
) -> list[dict]

Parameter	Type	Default	Description
workspace_id	str | None	None	Filter by workspace (defaults to client workspace)
context_id	str | None	None	Filter by context
include_expired	bool	False	Include expired sessions

set_session() / get_session_id() / clear_session()

def set_session(session_id: str) -> None
def get_session_id() -> str | None
def clear_session() -> None

These are synchronous helpers that manage the active session ID on the client. When set, the session ID is sent as the X-Session-ID header on all requests.

set_context()

Store key-value pair in session working memory.

async def set_context(
    session_id: str,
    key: str,
    value: Any,
    ttl_seconds: int | None = None,
) -> None

Parameter	Type	Default	Description
session_id	str	required	Session ID
key	str	required	Context key
value	Any	required	JSON-serializable value
ttl_seconds	int | None	None	Optional TTL for this key

get_context()

Retrieve values from session working memory.

async def get_context(
    session_id: str,
    keys: list[str],
) -> dict

commit_session()

Commit working memory to long-term storage.

async def commit_session(
    session_id: str,
    min_importance: float = 0.5,
    deduplicate: bool = True,
    categories: list[str] | None = None,
    max_memories: int = 50,
) -> dict

Parameter	Type	Default	Description
session_id	str	required	Session ID
min_importance	float	0.5	Minimum importance threshold
deduplicate	bool	True	Deduplicate memories
categories	list[str] | None	None	Filter by categories
max_memories	int	50	Maximum memories to extract

Returns: Dict with memories_created, memories_extracted, memories_deduplicated.

touch_session()

Extend session TTL.

async def touch_session(session_id: str) -> dict

Returns: Dict with updated expires_at.

delete_session()

async def delete_session(session_id: str) -> bool

---

Workspace Operations

create_workspace()

async def create_workspace(name: str) -> Workspace

get_workspace()

async def get_workspace(workspace_id: str | None = None) -> Workspace

Uses the client’s default workspace_id if not provided.

update_workspace()

async def update_workspace(
    workspace_id: str,
    name: str | None = None,
    settings: dict | None = None,
) -> Workspace

get_workspace_schema()

async def get_workspace_schema(workspace_id: str) -> dict

Returns: Dict with relationship_types, memory_subtypes, and can_customize.

create_context()

async def create_context(
    workspace_id: str,
    name: str,
    description: str | None = None,
    settings: dict | None = None,
) -> dict

list_contexts()

async def list_contexts(workspace_id: str) -> list[dict]

export_workspace()

Export workspace memories and associations as JSON.

async def export_workspace(
    workspace_id: str | None = None,
    include_associations: bool = True,
    offset: int = 0,
    limit: int = 0,
) -> dict

Returns: Dict with version, workspace_id, memories, associations, total_memories, total_associations.

export_workspace_stream()

Export as streaming NDJSON (for large workspaces).

async def export_workspace_stream(
    workspace_id: str | None = None,
    include_associations: bool = True,
    offset: int = 0,
    limit: int = 0,
) -> AsyncGenerator[dict, None]

Yields: Parsed JSON objects with type field ("header", "memory", "association", "footer").

import_workspace()

async def import_workspace(workspace_id: str, data: dict) -> dict

Returns: Dict with imported, skipped_duplicates, errors.

import_workspace_stream()

Import from NDJSON format.

async def import_workspace_stream(
    workspace_id: str,
    ndjson_lines: list[dict],
) -> dict

---

Briefing

get_briefing()

async def get_briefing(
    lookback_hours: int | None = None,
    lookback_minutes: int = 60,
    detail_level: str = "abstract",
    limit: int = 10,
    include_memories: bool = True,
    include_contradictions: bool = True,
) -> SessionBriefing

Parameters:

Parameter	Type	Default	Description
lookback_hours	int | None	None	Deprecated. Use lookback_minutes.
lookback_minutes	int	60	Time window in minutes
detail_level	str	"abstract"	"abstract", "overview", or "full"
limit	int	10	Max recent memories to include
include_memories	bool	True	Include memory content
include_contradictions	bool	True	Flag contradicting memories

Returns: SessionBriefing with fields: workspace_summary, recent_activity, open_threads, contradictions_detected.

---

Chat Thread Operations

create_thread()

async def create_thread(
    *,
    workspace_id: str | None = None,
    thread_id: str | None = None,
    user_id: str | None = None,
    context_id: str | None = None,
    observer_id: str | None = None,
    subject_id: str | None = None,
    title: str | None = None,
    metadata: dict | None = None,
    expires_at: str | None = None,
) -> ChatThread

list_threads()

async def list_threads(
    *,
    workspace_id: str | None = None,
    user_id: str | None = None,
    limit: int = 50,
    offset: int = 0,
) -> list[ChatThread]

get_thread()

async def get_thread(thread_id: str, *, workspace_id: str | None = None) -> ChatThread

get_thread_full()

Get thread with messages inlined.

async def get_thread_full(
    thread_id: str,
    *,
    workspace_id: str | None = None,
    limit: int = 100,
    offset: int = 0,
    order: str = "asc",
) -> ChatThreadWithMessages

Returns: ChatThreadWithMessages with fields: thread, messages, total_messages.

delete_thread()

async def delete_thread(thread_id: str, *, workspace_id: str | None = None) -> None

append_messages()

async def append_messages(
    thread_id: str,
    messages: list[dict],
    *,
    workspace_id: str | None = None,
) -> list[ChatMessage]

Each message dict should have: role (str), content (str), and optionally metadata (dict).

get_messages()

async def get_messages(
    thread_id: str,
    *,
    workspace_id: str | None = None,
    limit: int = 100,
    offset: int = 0,
    after_index: int | None = None,
    order: str = "asc",
) -> list[ChatMessage]

decompose_thread()

Trigger memory decomposition for unprocessed messages.

async def decompose_thread(
    thread_id: str,
    *,
    workspace_id: str | None = None,
) -> DecompositionResult

Returns: DecompositionResult with fields: thread_id, workspace_id, messages_processed, memories_created, from_index, to_index.

---

Context Environment Operations

All context environment methods require an active session (call set_session() or create_session() first).

context_exec()

Execute Python code in the session’s sandbox.

async def context_exec(
    code: str,
    result_var: str | None = None,
    return_result: bool = True,
    max_return_chars: int = 10_000,
) -> dict

Parameter	Type	Default	Description
code	str	required	Python code to execute
result_var	str | None	None	Store expression result in this variable
return_result	bool	True	Include result in response
max_return_chars	int	10000	Max chars for result serialization

Returns: Dict with output, result, error, variables_changed.

context_inspect()

Inspect sandbox state or a specific variable.

async def context_inspect(
    variable: str | None = None,
    preview_chars: int = 200,
) -> dict

context_load()

Load memories into the sandbox as a variable.

async def context_load(
    var: str,
    query: str,
    limit: int = 50,
    types: list[str] | None = None,
    tags: list[str] | None = None,
    min_relevance: float | None = None,
    include_embeddings: bool = False,
) -> dict

Returns: Dict with count and variable info.

context_inject()

Inject a value into the sandbox state.

async def context_inject(
    key: str,
    value: Any,
    parse_json: bool = False,
) -> dict

context_query()

Send sandbox variables and a prompt to the server-side LLM.

async def context_query(
    prompt: str,
    variables: list[str],
    max_context_chars: int | None = None,
    result_var: str | None = None,
) -> dict

Returns: Dict with response and tokens_used.

context_rlm()

Run a Recursive Language Model (RLM) reasoning loop.

async def context_rlm(
    goal: str,
    memory_query: str | None = None,
    memory_limit: int = 100,
    max_iterations: int = 10,
    variables: list[str] | None = None,
    result_var: str | None = None,
    detail_level: str = "standard",
) -> dict

Parameter	Type	Default	Description
goal	str	required	Natural language goal
memory_query	str | None	None	Optional query to load initial data
memory_limit	int	100	Max memories to load
max_iterations	int	10	Max reasoning iterations
variables	list[str] | None	None	Variables to include in context
result_var	str | None	None	Store final result in this variable
detail_level	str	"standard"	"brief", "standard", or "detailed"

Returns: Dict with result, iterations, and execution trace.

context_status()

async def context_status() -> dict

Returns: Dict with variable_count, memory_bytes, and metadata.

context_checkpoint()

Checkpoint sandbox state for persistence (enterprise deployments).

async def context_checkpoint() -> None

context_cleanup()

Clean up and remove the sandbox environment.

async def context_cleanup() -> None

---

SyncMemoryLayerClient

Synchronous wrapper with identical methods (no async/await). Use as a regular context manager.

from memorylayer import SyncMemoryLayerClient

with SyncMemoryLayerClient(
    base_url="http://localhost:61001",
    api_key="your-key",
    workspace_id="ws_123"
) as client:
    memory = client.remember("User prefers Python")
    results = client.recall("coding preferences")

The constructor accepts the same parameters as MemoryLayerClient. All methods have the same signatures, minus async/await. The sync client also exposes connect() and close() for manual lifecycle management.

sync_client()

Convenience context manager function:

from memorylayer import sync_client

with sync_client(
    base_url="http://localhost:61001",
    api_key="your-key",
    workspace_id="ws_123"
) as client:
    memory = client.remember("Hello world")

---

Types

MemoryType

class MemoryType(str, Enum):
    EPISODIC = "episodic"      # Specific events/interactions
    SEMANTIC = "semantic"      # Facts, concepts, relationships
    PROCEDURAL = "procedural"  # How to do things
    WORKING = "working"        # Current task context

MemorySubtype

class MemorySubtype(str, Enum):
    SOLUTION = "solution"
    PROBLEM = "problem"
    CODE_PATTERN = "code_pattern"
    FIX = "fix"
    ERROR = "error"
    WORKFLOW = "workflow"
    PREFERENCE = "preference"
    DECISION = "decision"
    DIRECTIVE = "directive"
    PROFILE = "profile"
    ENTITY = "entity"
    EVENT = "event"
    INFERENCE = "inference"

RecallMode

class RecallMode(str, Enum):
    RAG = "rag"        # Active: vector similarity search + graph traversal
    LLM = "llm"        # Deprecated: query rewriting + LLM-driven retrieval
    HYBRID = "hybrid"  # Deprecated: RAG-first, fall back to LLM

LLM and HYBRID are deprecated. Use RAG for recall and the Context Environment with context_rlm() when you need LLM-driven analysis over memories.

SearchTolerance

class SearchTolerance(str, Enum):
    LOOSE = "loose"        # Fuzzy matching, broader results
    MODERATE = "moderate"  # Balanced precision/recall
    STRICT = "strict"      # Exact matching, high relevance

RelationshipType

class RelationshipType(str, Enum):
    # Causal
    CAUSES = "causes"
    TRIGGERS = "triggers"
    LEADS_TO = "leads_to"
    PREVENTS = "prevents"
    # Solution
    SOLVES = "solves"
    ADDRESSES = "addresses"
    ALTERNATIVE_TO = "alternative_to"
    IMPROVES = "improves"
    # Context
    OCCURS_IN = "occurs_in"
    APPLIES_TO = "applies_to"
    WORKS_WITH = "works_with"
    REQUIRES = "requires"
    # Learning
    BUILDS_ON = "builds_on"
    CONTRADICTS = "contradicts"
    CONFIRMS = "confirms"
    SUPERSEDES = "supersedes"
    # Similarity
    SIMILAR_TO = "similar_to"
    VARIANT_OF = "variant_of"
    RELATED_TO = "related_to"
    # Workflow
    FOLLOWS = "follows"
    DEPENDS_ON = "depends_on"
    ENABLES = "enables"
    BLOCKS = "blocks"
    # Quality
    EFFECTIVE_FOR = "effective_for"
    PREFERRED_OVER = "preferred_over"
    DEPRECATED_BY = "deprecated_by"

60+ relationship types organized into 11 categories:

• Hierarchical: parent_of, child_of, part_of, contains, instance_of, subtype_of
• Causal: causes, triggers, leads_to, prevents, and more
• Temporal: precedes, concurrent_with, follows_temporally
• Similarity: similar_to, variant_of, related_to, analogous_to
• Learning: builds_on, contradicts, confirms, supersedes, and more
• Refinement: refines, abstracts, specializes, generalizes
• Reference: references, referenced_by
• Solution: solves, addresses, alternative_to, improves, and more
• Context: occurs_in, applies_to, works_with, requires, and more
• Workflow: follows, depends_on, enables, blocks, and more
• Quality: effective_for, preferred_over, deprecated_by, and more

Use get_workspace_schema() to list all available relationship types.

RelationshipCategory

class RelationshipCategory(str, Enum):
    HIERARCHICAL = "hierarchical"
    CAUSAL = "causal"
    TEMPORAL = "temporal"
    SIMILARITY = "similarity"
    LEARNING = "learning"
    REFERENCE = "reference"
    SOLUTION = "solution"
    CONTEXT = "context"
    WORKFLOW = "workflow"
    QUALITY = "quality"

---

Model Classes

Memory

class Memory(BaseModel):
    id: str
    workspace_id: str
    content: str
    type: MemoryType
    subtype: MemorySubtype | None
    importance: float
    tags: list[str]
    metadata: dict
    access_count: int
    last_accessed_at: datetime | None
    created_at: datetime
    updated_at: datetime
    context_id: str | None
    user_id: str | None
    session_id: str | None
    decay_factor: float
    pinned: bool
    status: str
    abstract: str | None
    overview: str | None
    relevance_score: float | None
    # ... additional server fields

Association

class Association(BaseModel):
    id: str
    workspace_id: str
    source_id: str
    target_id: str
    relationship: str
    strength: float
    metadata: dict
    created_at: datetime

Session

class Session(BaseModel):
    id: str
    workspace_id: str
    user_id: str | None
    metadata: dict
    expires_at: datetime
    created_at: datetime

Workspace

class Workspace(BaseModel):
    id: str
    tenant_id: str
    name: str
    settings: dict
    created_at: datetime
    updated_at: datetime

ChatThread

class ChatThread(BaseModel):
    id: str
    workspace_id: str
    user_id: str | None
    context_id: str
    title: str | None
    metadata: dict
    message_count: int
    expires_at: datetime | None
    created_at: datetime
    updated_at: datetime

ChatMessage

class ChatMessage(BaseModel):
    id: str
    thread_id: str
    message_index: int
    role: str
    content: str | list[ChatMessageContent]
    metadata: dict
    created_at: datetime

---

Exceptions

MemoryLayerError              # Base exception
├── AuthenticationError       # 401 - Invalid API key
├── AuthorizationError        # 403 - Access denied
├── NotFoundError             # 404 - Resource not found
├── ValidationError           # 422 - Invalid request
├── RateLimitError            # 429 - Rate limit exceeded
├── ServerError               # 500+ - Server error
└── EnterpriseRequiredError   # 501 - Enterprise feature required

All exceptions have these properties:

Property	Type	Description
message	str	Error message
status_code	int	HTTP status code

EnterpriseRequiredError also has a feature property identifying which enterprise feature was requested.