Python SDK

Zero-dependency (stdlib only). Python ≥ 3.9.

Install#

# PyPI publish pending — install from source:
pip install "git+https://github.com/11data/longmem#subdirectory=sdk/python"

Constructor#

from longmem import Longmem

mem = Longmem(
    api_key=None,        # default: LONGMEM_API_KEY env var
    base_url=None,       # default: LONGMEM_BASE_URL env var or https://rememberos.ai
    timeout=30.0,        # seconds per request
    max_retries=2,       # retry transient failures (429/5xx/network); 0 disables
    backoff_base=0.5,    # seconds, doubles each retry (Retry-After header wins)
    backoff_cap=8.0,
)

Raises LongmemError immediately if no key is available.

Automatic retries#

Transient failures — network errors and 429/500/502/ 503/504 — are retried with exponential backoff, honoring a Retry-After header when present. 4xx responses are returned immediately (never retried). Tune with max_retries / backoff_base / backoff_cap, or set max_retries=0 to disable.

Methods#

Method	Returns
`add(text, *, collection="default", importance=0.7, category="other", metadata=None, source="sdk", container_tag=None)`	the created memory dict
`add_many(items, *, collection="default", container_tag=None, importance=0.7, category="other", source="sdk")`	`{"created", "ids"}`; bulk import, items are strings or dicts, chunked to 100/call
`remember(content, *, collection="default", sync=True)`	extracted facts (sync) or `{"job_id"}` (sync=False)
`update(memory_id, *, collection="default", text=None, importance=None, category=None, metadata=None)`	the updated memory; only set fields change, text re-embeds
`search(query, *, collection=None, limit=5, min_score=0.3, mode="hybrid", category=None, container_tag=None)`	list of result dicts; `collection=None` searches everything
`get(memory_id, *, collection="default")`	one memory dict
`list_memories(*, collection="default", limit=50, offset=0, include_archived=False)`	list of memories, newest first (pinned first)
`move(memory_id, *, collection="default", to_collection)`	`{moved}`; move a memory to another collection (created if needed)
`delete(memory_id, *, collection="default")`	`{}`; deletes the memory (+ its files)
`delete_collection(collection)`	`{}`; deletes the collection and all its memories
`forget(query, *, collection="default", min_score=0.5, limit=10)`	`{"deleted"}`; semantic delete — destructive
`collections()`	list of collection dicts
`profile(collection)`	`{"profile", "based_on"}`
`related(memory_id, *, collection="default", limit=5, min_score=0.0)`	list of related memories (nearest neighbours, excludes self/archived)
`dedup(*, collection="default", threshold=0.95, dry_run=True)`	`{clusters, deleted, dry_run}`; consolidate near-duplicates
`contradictions(*, collection="default", limit=100)`	list of `{current, superseded}` pairs
`pin(memory_id, *, collection="default", pinned=True)`	the updated memory; pin/unpin
`archive(memory_id, *, collection="default", archived=True)`	the updated memory; archive/restore
`archive_stale(*, collection="default", older_than_days=90, dry_run=True)`	`{would_archive}` or `{archived}`
`health(*, collection="default", stale_days=90)`	collection health snapshot — `total`, `stale`, `expired`, `supersessions`, `needs_attention`, `categories`
`summary(*, collection="default", limit=50)`	board-report briefing — `{summary, themes, based_on, generated_at}`
`create_connection(provider, *, name=None, collection="default", container_tag=None, config=None, credentials=None)`	the connection; managed connector synced server-side (creds encrypted, never returned)
`list_connections()` / `get_connection(id)`	connection dicts (no credentials)
`sync_connection(id)`	`{synced, cursor, status}` or `{status:"error", error}`
`connection_runs(id, *, limit=20)`	recent sync runs: `{status, synced_count, error, created_at}`
`delete_connection(id)`	`{}`; removes the connection (synced memories kept)
`connector_health(*, hours=24)`	admin: cross-tenant sync health `[{provider, runs, errors, last_run}]`
`connector_status()`	admin: connection counts by status `[{status, n}]` — surfaces paused/erroring connectors
`admin_memory_health()`	admin: fleet-wide memory hygiene `{total, current, archived, stale, expired, supersessions, needs_attention}`
`export_account()`	full knowledge-base dump `{tenant, collections, memories, …}` — text-only, re-imports anywhere
`import_memories(memories, *, collections=None, into_collection=None)`	re-ingest a dump (re-embeds text); `{imported, skipped, collections}`

# read, delete, and semantic-forget
m = mem.get(memory_id, collection="notes")
mem.delete(memory_id, collection="notes")
mem.forget("everything about project Icarus", collection="notes", min_score=0.6)  # destructive
mem.delete_collection("scratch")               # drops the whole collection

Bulk import#

# strings, or dicts for per-item fields; chunked to 100/call, retries per chunk
mem.add_many([
    "Alex prefers dark mode",
    {"text": "Bianca is on the design team", "container_tag": "user-42", "metadata": {"src": "crm"}},
], collection="memories")   # -> {"created": 2, "ids": [...]}

Errors#

from longmem import Longmem, LongmemError, LongmemAPIError

try:
    mem.add("…")
except LongmemAPIError as e:    # non-2xx from the API
    print(e.status, e)          # e.g. 403 read-only key
except LongmemError as e:       # config / network
    print(e)

A complete agent loop#

from longmem import Longmem

mem = Longmem()
COLLECTION = "assistant"

def respond(user_id: str, message: str) -> str:
    context = mem.search(message, collection=COLLECTION,
                         container_tag=f"user-{user_id}", limit=5)
    answer = llm(message, context=[h["text"] for h in context])
    mem.remember(message, collection=COLLECTION)   # async; facts evolve
    return answer

PreviousAPI Reference NextJavaScript SDK