Python - every method, three groups.
Write, read, delete. Every example below was run against the live API on 2026-08-01; responses are verbatim.
pip install wontopos
from wontopos import Client mem = Client(api_key="wos-live-...") # or read from an env var
from wontopos import AsyncClient # pip install "wontopos[async]" async with AsyncClient(api_key="wos-live-...") as mem: hits = await mem.search("what does she drink?", user_id="alice")
AsyncClient has the same methods, awaited; install it with pip install "wontopos[async]". Client is safe after fork(): the child drops the connections it inherited. An AsyncClient made before fork() builds its own connections in the child.
Choose a model
The API key picks which memory (your workspace); the model picks which engine reads it. Every model shares one memory, so you can store with one and recall with another. Set a default on the client; override a single call by passing model=.
mem = Client(api_key="wos-live-...") # default engine: tablet-2 mem.recall("...", user_id="alice") # tablet-2 mem.recall("...", user_id="alice", model="tablet-1") # or pick a model per call
list_models
The catalog - the ids you can pass to model and whether each is live. memory: "shared" models read the same store; "isolated" keeps its own. Needs no API key.
mem.list_models()[{"id": "tablet-1", "name": "Tablet 1", "available": true, "memory": "shared"},
{"id": "tablet-2", "name": "Tablet 2", "available": true, "memory": "shared"},
{"id": "scroll-1", "name": "Scroll 1", "available": true, "memory": "shared"},
{"id": "scroll-1.2", "name": "Scroll 1.2", "available": true, "memory": "shared"}]ping
Confirm connectivity and that your API key works - a one-line setup check.
mem.ping() # True, or raises AuthenticationError / PaymentRequiredError
The catalog above always reflects the models available right now - pass any other id and you get a clear error. New models appear there automatically when they ship.
Write
add
Store one memory. No LLM call on the way in - you pay the write rate and nothing else.
mem.add("she prefers tea over coffee", user_id="alice") mem.add("I promised the summary by Friday", user_id="alice", speaker="me") # its own words - no registration needed
{"id": "576700aa-f0e0-4c26-99a0-10e2d5b0d624", "status": "stored (1 chunks)"}add refuses a metadata key that spells a store id or the idempotency key, such as userId, store_id or idempotencyKey, in any case and with _, - or . anywhere. It raises ValueError and sends nothing, for metadata= and extra keyword arguments alike. Pass the store as user_id= and the key as idempotency_key=. The service keeps four metadata keys, speaker, event_date, category and conversation_id, and drops any other; the SDK warns once for each key it drops.add_turn
Store one conversation turn (user + assistant) into short-term and long-term memory at once.
mem.add_turn("hi", "hello!", user_id="alice")
{"status": "ok"}speaker
Every memory can carry who said it. Register a person once, then pass their name as the speaker; "me" (the assistant's own words) never needs registration. Search accepts a speaker too, so you can recall one person's words only.
mem.add_speaker("Bob", user_id="alice") # once per person; "me" needs no registration mem.add("I promised to send the report on Friday", user_id="alice", speaker="me") mem.add("Bob said the deadline moved to Tuesday", user_id="alice", speaker="Bob") mem.search("what did Bob say about deadlines?", user_id="alice", speaker="Bob")
[{"content": "Bob said the deadline moved to Tuesday", "speaker": "Bob", ...}]add_bulk
Backfill a large blob of text. Split and indexed server-side - ideal for importing existing history.
mem.add_bulk("Alice moved to Brooklyn in March. She works at a design studio downtown.", user_id="alice") mem.add_bulk(history, user_id="alice", timestamp="2024-03-01T10:00:00Z") # date a backfill (RFC3339)
{"elapsed_secs": 0.154154944, "status": "ok", "stored": 1, "total_chunks": 1}update
A fact changed. The old memory is marked superseded (kept for history); the new one takes its place in recall.
mem.update("576700aa-...", "she switched to coffee this year", user_id="alice")
{"new_memory_id": "07e94433-b7cc-4e49-8d8f-f37fc1a392b7",
"old_memory_id": "576700aa-f0e0-4c26-99a0-10e2d5b0d624", "status": "superseded"}Read
search
Semantic search, most relevant first. Any language finds any memory, whichever language it was written in. The SDK returns one list: the memories, then the agent's own words (speaker "me"), then any image rows the answer carried, with no id repeated. limit (5 to 20, default 10) sets how many memories come back. The other rows come on top, so size your prompt on what you receive. search_full returns memories, self_memories, images and search_ms as separate fields.
r = mem.search("what does she drink?", user_id="alice", limit=5)
[{
"id": "576700aa-f0e0-4c26-99a0-10e2d5b0d624",
"content": "she prefers tea over coffee",
"category": "general",
"time_bucket": "2026-06",
"importance": 0.3,
"similarity": 0.6316057443618774,
"is_superseded": false,
"superseded_by": null,
"created_at": "2026-07-10T04:20:39.688276876Z"
}]| Field | Meaning |
|---|---|
| similarity | How close this memory is to your query (0–1). |
| is_superseded | True if this fact was replaced by update(). |
| search_ms | Server-side retrieval time. It is in the HTTP body and in search_full. |
recall
One round-trip returns everything your LLM needs - paste the result straight into your prompt: the recent turns, at most limit memories and at most context_limit rows of the surrounding conversation, however much you've stored.
ctx = mem.recall("what does she drink?", user_id="alice")
{"short_term": {"count": 2, "turns": [{"role": "user", "content": "hi", ...}]},
"long_term": {"count": 4, "memories": [{"content": "she prefers tea over coffee",
"similarity": 0.63, ...}]},
"context": {"count": 4, "around_top_memory": [
"[match] she prefers tea over coffee",
"[after] Alice moved to Brooklyn in March. ..."]},
"instruction": "Use short_term for recent context, long_term for relevant
past memories, context for surrounding conversation of the
most relevant memory."}history
Recent conversation turns (short-term memory), oldest first.
turns = mem.history("alice")
{"count": 2, "turns": [
{"role": "user", "content": "hi", "timestamp": "2026-07-10T04:20:40.989011337Z"},
{"role": "assistant", "content": "hello!", "timestamp": "2026-07-10T04:20:40.989013416Z"}
], "user_id": "alice"}stats
Memory counts for one user.
mem.stats("alice")
{"short_term_turns": 2, "total_memories": 4, "user_id": "alice"}usage
What this key has spent and what is left. Free, and it skips the balance gate — an account at zero must still be able to find out why.
mem.usage(days=7)
{"window_days": 7, "key": {"requests": 128, "cost_cents": 41.9, "since": "lifetime"}, "stores": [{"store": "alice", "cost_cents": 12.4}], "balance_cents": 1840}get
Fetch one memory by id - the id that add or list_memories returned. The original text plus its metadata, and nothing internal. An id from another store, or one that was deleted, returns 404.
m = mem.get("alice", memory_id="576700aa-...")
{"id": "576700aa-...", "content": "she prefers tea over coffee",
"category": "general", "created_at": "2026-07-10T04:20:39Z", "event_date": null,
"is_superseded": false, "superseded_by": null}get needs SDK 2.2.40 or later. Earlier versions returned an empty result there.list_memories
List a store's memories - the original text you stored plus its metadata, and nothing internal. Paged by cursor: pass the returned next_cursor back for the next page.
page = mem.list_memories("alice", limit=100)
{"count": 2, "next_cursor": null, "memories": [
{"id": "576700aa-...", "content": "she prefers tea over coffee",
"category": "general", "created_at": "2026-07-10T04:20:39Z", "event_date": null,
"is_superseded": false, "superseded_by": null}
]}iter_memories · export_memories
Page through every memory with no cursor bookkeeping, or pull the whole store at once.
for m in mem.iter_memories("alice"): # every page, no cursor bookkeeping print(m["id"], m["content"]) everything = mem.export_memories("alice") # the whole store as a list
iter_images · export_images
The same pair for image memories. Each row describes one image; fetch its bytes with get_image. export_images is new in 2.2.39, and page_size (5 to 20) sets how many rows each request fetches.
for img in mem.iter_images("alice"): # one image row at a time print(img["id"]) images = mem.export_images("alice") # every image row as a list
Delete
delete
Delete a single memory by id.
mem.delete("alice", memory_id="576700aa-...")
{"memory_id": "576700aa-f0e0-4c26-99a0-10e2d5b0d624", "status": "deleted"}delete_all
Erase everything for one user - one call, GDPR-ready.
mem.delete_all("alice")
{"memories_deleted": 4, "status": "deleted", "user_id": "alice"}Errors and reliability
Every failure is a typed error - catch the specific one (rate limit, auth, payment) or all of them with the base WosError.
from wontopos import PaymentRequiredError, NotFoundError try: mem.add("...", user_id="alice") except NotFoundError: mem.create_store("alice") # store didn't exist yet except PaymentRequiredError: top_up() # out of credit - don't retry
ValueError before anything is sent: a limit, page size or max_images out of range, a blank user_id, an unknown search option, or a metadata key that names a store. except WosError does not catch it. Fix the call; retrying it will not help.rate_limit
Read the quota left after any call and slow down before you hit the limit.
mem.search("...", user_id="alice") rl = mem.rate_limit # {"limit": 150, "remaining": 3, "reset": ...}
search_self
Both lanes from one call on a self-memory model (Scroll 1.2+): what others said, and the agent's OWN words - kept apart so the reader never confuses who said what.
r = mem.search_self("what did I promise?", user_id="alice") r["memories"] # what others said / general memories r["self_memories"] # the agent's OWN words (speaker "me")
list_engrams
Ask the service which engrams and delivery forms the selected model can run, instead of hard-coding names that go stale the moment a new one ships.
cat = mem.list_engrams() [e["name"] for e in cat["engrams"]] # ask, never hard-code
filters
Narrow the search to part of a store. A narrow filter can return fewer than limit even when the store holds more inside it.
mem.search("what did we decide", user_id="alice", filters={ "categories": ["work"], "event_from": "2026-01-01", # when it HAPPENED })
YYYY-MM-DD, and a plain end date covers that whole day. Filters apply to the memories, not to the agent's own words (self_memories). The service drops a key it does not know instead of refusing it, so a typo silently widens the search. The SDKs print a warning the first time they see an unknown filter key.speaker, filters, verify, max_images, cache_control, form and tz. Any other option is refused before sending, and the error names the option you probably meant. Put a field the service accepts but this version does not know under extra. form and tz were refused in 2.2.38 and 2.2.39 and work again from 2.2.40.idempotency_key
Makes repeating one write safe. Use it when the retry is yours - a job that died and was re-run, a queue that redelivers.
Every memory write already carries a key: when you pass none, the client makes one for the call, so its own retries of add, add_turn, add_bulk and update cannot store twice. Pass your own when the retry is yours. The service keeps a key for 24 hours. A repeat that arrives while the first request is still running is told to wait and is sent again; one whose first request did not come back with an answer is refused with a 409, and you check before sending it again under a new key.
mem.add("she prefers tea", "alice", idempotency_key=f"import:{row.id}")
import:row-42), never a constant: one key reused for two different writes replays the first, and the second is silently lost. Format: 1-128 chars of [A-Za-z0-9._:-].with_timeout / with_retries / with_deadline
Tune one call site without touching the client you already built: a clone with a longer timeout for a big backfill, or with retries off inside your own retry loop.
timeout bounds one attempt, so a call that retries can outlive it: at the defaults a single call can hold a connection for 30s, back off, try again, and again. deadline bounds the whole call instead - every attempt is capped at what is left, and a backoff is never slept past the budget. Set it when the caller has a real limit, like a request handler with five seconds. When a retry would not fit in the deadline, the call raises the error of the answer it got instead of a deadline error.
By default a failed call is tried again up to twice. A 429 is retried on every call, after Retry-After. 408, 502, 503 and 504 are retried on calls that only read, on the DELETE calls and on the memory writes, such as search, recall, get, list_memories, ping, list_stores, usage, delete_store, add and update: a memory write carries an idempotency key, so a repeat is answered with the first result. The other writes sent as POST (create_store, add_speaker, delete, delete_all) get those statuses back on the first failure. A connection that could not be opened is retried on every call. A 409 for a write already in flight, or for a key whose first request is still running, is retried on every call, after the wait the service asks for. A Retry-After over 30 seconds is not waited out: the call raises at once. A timeout after the request went out is not retried. The SDKs also retry a read, a DELETE call or a memory write when the connection drops while the reply is being read. Another write sent as POST is not retried then, because the first attempt may already have landed.
mem.with_timeout(120).add_bulk(big_blob, "alice") # this slow call only mem.with_retries(0).add("...", "alice") # you retry, not the SDK mem.with_deadline(5).recall("...", "alice") # 5s for the whole call