Memory class: the OMP synchronous client

The Memory class is the synchronous OMP client. You instantiate it once with a provider string and any provider-specific configuration, then call its methods — add, search, get, update, delete, list, context, audit, and capabilities — the same way regardless of which backend is underneath. Swapping providers requires only changing the constructor arguments; all method calls remain identical.

from openmem import Memory

Constructor

Memory(provider="postgres", **config)

The provider argument is a string that selects the backend adapter. The remaining keyword arguments are forwarded directly to that adapter.

provider

string

default:"postgres"

required

Selects the memory backend. Accepted values: "postgres", "mem0", "supermemory", "letta". Pass base_url= pointing to any OMP-native HTTP server and the SDK will auto-detect it as a passthrough provider regardless of this string.

The additional **config keyword arguments depend on which provider you choose:

postgres

url

string

required

PostgreSQL connection string, e.g. "postgresql://localhost/omp". Also accepted as dsn=.

embedder

object

An embedder instance. Defaults to a lightweight offline embedder that uses zero vectors — suitable for testing without an OpenAI key. Pass use_openai=True to use OpenAI embeddings instead, or supply a custom embedder object.

use_openai

boolean

default:"false"

When True and no embedder= is supplied, initializes an OpenAIEmbedder automatically.

mem0

api_key

string

required

Your Mem0 API key.

host

string

default:"https://api.mem0.ai"

Mem0 API base URL. Override for self-hosted or staging environments.

supermemory

api_key

string

required

Your Supermemory API key.

base_url

string

Override the default Supermemory base URL.

letta

api_key

string

required

Your Letta API key.

base_url

string

Override the default Letta base URL.

passthrough (any OMP-native server)

base_url

string

required

Base URL of the OMP-native server. When this is provided, the SDK probes /capabilities. If the response contains omp_version, a passthrough adapter is used and the provider string is ignored.

api_key

string

API key forwarded as a Bearer token.

Methods

add()

Store a new memory for a user.

mem.add(
    content,
    user_id,
    scope=None,
    tags=None,
    source=None,
    confidence=None,
    valid_from=None,
    valid_to=None,
    supersedes=None,
) → MemoryRecord

content

string

required

The text content of the memory. Must be a non-empty string.

user_id

string

required

Identifies the user this memory belongs to. All subsequent reads and searches are scoped to this user.

scope

string

Slash-delimited hierarchy that organizes the memory, e.g. "coding/preferences" or "health/symptoms". Scopes enable fine-grained consent and filtering.

search()

Find memories semantically relevant to a query string.

mem.search(query, user_id, scope=None, limit=10, min_score=None) → list[SearchResult]

query

string

required

The natural-language query to search against stored memories.

user_id

string

required

Restricts results to memories belonging to this user.

scope

string

Restrict results to memories under this scope prefix, e.g. "coding/*".

limit

number

default:"10"

Maximum number of results to return. Subject to the provider’s limits.max_search_results cap.

min_score

number

Minimum similarity score (0..1) for a result to be included. Omit to return the top limit results regardless of score.

Returns a list of SearchResult objects, each containing a memory and a score.

get()

Fetch a single memory by its provider-assigned ID.

mem.get(id) → MemoryRecord

string

required

The provider-assigned memory ID, e.g. "mem_abc123".

Returns a MemoryRecord. Raises NotFoundError if the ID does not exist.

update()

Partially update a memory’s fields. Only the fields you provide are changed.

mem.update(
    id,
    content=None,
    scope=None,
    tags=None,
    confidence=None,
    valid_to=None,
    supersedes=None,
) → MemoryRecord

string

required

The ID of the memory to update.

content

string

Replacement text content.

scope

string

New scope path to move the memory to.

delete()

Permanently delete a memory.

mem.delete(id) → None

string

required

The ID of the memory to delete.

Returns None on success. Raises NotFoundError if the ID does not exist.

list()

List memories for a user, optionally filtered by scope, tag, or time range. Results are paginated.

mem.list(
    user_id,
    scope=None,
    tag=None,
    since=None,
    until=None,
    limit=50,
    cursor=None,
) → MemoryPage

user_id

string

required

The user whose memories to list.

scope

string

Filter to memories under this scope prefix.

tag

string

Filter to memories with this exact tag.

since

datetime

Return only memories created at or after this datetime.

until

datetime

Return only memories created before this datetime.

limit

number

default:"50"

Maximum number of records per page.

cursor

string

Opaque pagination cursor from a previous MemoryPage.next_cursor. Pass None to start from the beginning.

Returns a MemoryPage. When next_cursor on the result is non-None, pass it back as cursor= to fetch the next page.

context()

Retrieve a pre-ranked, prompt-ready text block for injecting into an LLM prompt.

mem.context(query, user_id, scope=None, token_budget=500) → ContextBlock

query

string

required

The user’s current query or intent used to rank memories for relevance.

user_id

string

required

The user whose memories to draw from.

scope

string

Restrict context retrieval to memories under this scope.

token_budget

number

default:"500"

Approximate maximum number of tokens for the returned text. The provider fits as many high-scoring memories as possible within this budget.

Returns a ContextBlock with a text field ready for prompt injection, citations linking back to source memories, and an optional token_count.

audit()

Retrieve the audit log of operations performed on a user’s memories.

mem.audit(user_id, app=None, since=None, limit=100) → list[AuditEntry]

Audit log support depends on the provider. Check caps.features.supports_audit before calling this method.

user_id

string

required

The user whose audit log to retrieve.

app

string

Filter to entries originating from this app name.

since

datetime

Return only entries at or after this datetime.

limit

number

default:"100"

Maximum number of entries to return.

Returns a list of AuditEntry objects, each describing a single operation.

capabilities()

Retrieve the provider’s declared capabilities.

mem.capabilities() → Capabilities

Returns a Capabilities object describing which verbs and features the provider supports. The result is cached for the lifetime of the Memory instance — the network call is made at most once per session. Use this to conditionally enable features at runtime:

caps = mem.capabilities()
if caps.features.temporal:
    results = mem.search("last week's work", user_id=uid)
if caps.features.supports_audit:
    log = mem.audit(user_id=uid)

Full example

from datetime import datetime, timezone
from openmem import Memory

mem = Memory(provider="postgres", url="postgresql://localhost/omp")

# Check what this provider supports
caps = mem.capabilities()
print(f"Provider: {caps.provider}, OMP version: {caps.omp_version}")

# Add a memory
record = mem.add(
    content="User prefers pnpm over npm",
    user_id="u1",
    scope="coding/preferences",
    tags=["tooling", "nodejs"],
    source={"app": "cursor", "type": "explicit"},
    confidence=0.95,
)
print(f"Stored: {record.id}")

# Search
results = mem.search(
    query="package manager preferences",
    user_id="u1",
    scope="coding/preferences",
    limit=5,
)
for r in results:
    print(f"{r.score:.2f}  {r.memory.content}")

# Fetch a specific memory
fetched = mem.get(record.id)
print(fetched.content)

# Update it
updated = mem.update(
    record.id,
    content="User prefers bun for new projects",
    supersedes=[record.id],
)

# Get prompt-ready context
ctx = mem.context(
    query="set up a new Node project",
    user_id="u1",
    scope="coding/preferences",
    token_budget=400,
)
prompt = f"Relevant memory:\n{ctx.text}\n\nUser: ..."

# List all memories (paginated)
page = mem.list(user_id="u1", scope="coding/preferences", limit=20)
for m in page.items:
    print(m.id, m.content)
if page.next_cursor:
    next_page = mem.list(user_id="u1", cursor=page.next_cursor, limit=20)

# Audit log
if caps.features.supports_audit:
    entries = mem.audit(user_id="u1", limit=10)
    for e in entries:
        print(e.action, e.memory_id, e.timestamp)

# Delete
mem.delete(updated.id)

AsyncMemory has identical method signatures to Memory, but every method is a coroutine and must be awaited. See AsyncMemory.

Get Started

Core Concepts

Guides

SDK Reference

Memory class: the OMP synchronous client

Constructor

Methods

add()

search()

get()

update()

delete()

list()

context()

audit()

capabilities()

Full example

Get Started

Core Concepts

Guides

SDK Reference

Documentation Index

​Constructor

​Methods

​add()

​search()

​get()

​update()

​delete()

​list()

​context()

​audit()

​capabilities()

​Full example

Constructor

Methods

add()

search()

get()

update()

delete()

list()

context()

audit()

capabilities()

Full example