Python SDK

Python - 全メソッドを 3 グループで。

書き込み、読み取り、削除。以下の例はすべて 2026-08-01 に本番 API に対して実行したもので、レスポンスは原文のままです。

pip install wontopos

from wontopos import Client

mem = Client(api_key="wos-live-...")  # or read from an env var

モデルを選ぶ

API キーはどの記憶か(あなたのアカウント)を決め、モデルはどのエンジンが読むかを決めます。すべてのモデルはひとつの記憶を共有するため、あるモデルで保存して別のモデルで呼び出せます。クライアントにデフォルトを設定し、個別の呼び出しは model= を渡して上書きします。

mem = Client(api_key="wos-live-...", model="tablet-1")  # default engine
mem.recall("...", user_id="alice")                  # tablet-1
mem.recall("...", user_id="alice", model="scroll-1")  # or pick a model per call

list_models

カタログ - model に渡せる id と、それぞれの提供状況です。memory: "shared" のモデルは同じストアを読み、"isolated" は独自のストアを持ちます。API キーは不要です。

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

接続と API キーが有効かを一行で確認します。

mem.ping()   # True, or raises AuthenticationError / PaymentRequiredError

上のカタログは、常に現時点で利用できるモデルを反映しています - それ以外の id を渡すと明確なエラーが返ります。新しいモデルは出荷と同時に自動で現れます。

書き込み

add

記憶を 1 件保存します。取り込みに LLM 呼び出しはなく、書き込み分の料金だけです。

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_turn

会話ターン 1 件(ユーザー + アシスタント)を、短期記憶と長期記憶に同時に保存します。

mem.add_turn("hi", "hello!", user_id="alice")

実際のレスポンス

{"status": "ok"}

speaker

すべての記憶に、誰の発言かを載せられます。人は一度登録し、その後は名前を speaker として渡してください。"me"（アシスタント自身の言葉）は登録不要です。検索にも speaker を指定すれば、その人の発言だけを取り出せます。

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")

response

[{"content": "Bob said the deadline moved to Tuesday", "speaker": "Bob", ...}]

話者はストアと同じく明示的です。先に人を登録し、その名前で保存します。タイプミスが静かに新しい人になることはありません。ストアあたりまず 50 人まで登録でき（順次拡大予定）、"me" は登録も数えられることもありません。

add_bulk

長いテキストを一度に取り込みます。サーバー側で分割して索引付けします - 既存の履歴の移行に適しています。

mem.add_bulk("Alice moved to Brooklyn in March. She works at a design studio downtown.", user_id="alice")

実際のレスポンス

{"elapsed_secs": 0.154154944, "status": "ok", "stored": 1, "total_chunks": 1}

update

事実が変わったとき。古い記憶は superseded とマークされ(履歴として保持)、新しい記憶がリコールでその座を引き継ぎます。

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"}

読み取り

search

セマンティック検索で、関連度の高い順に返します。どの言語で尋ねても、どの言語で書かれた記憶にも届きます。 SDK は memories 配列を直接返します。以下は生の HTTP ボディです。モデルによっては結果のまとまりを複数返し、SDK がそれを統合して返すため、配列が max_results より多くなることがあります。要求した数ではなく、受け取った配列を基準にプロンプトの大きさを決めてください。

r = mem.search("what does she drink?", user_id="alice", limit=1)

実際のレスポンス(HTTP ボディ)

[{
   "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"
 }]

フィールド	意味
similarity	この記憶がクエリにどれだけ近いか (0–1)。
is_superseded	update() によって置き換えられた事実なら true。
search_ms	サーバー側の検索所要時間。

recall

1 回の往復で、LLM に必要なものがすべて返ります - 結果をそのままプロンプトに貼るだけです。どれだけ保存していても、上限のある固定サイズのコンテキストです。

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

直近の会話ターン(短期記憶)を、古い順に返します。

turns = mem.history("alice")

実際のレスポンス(HTTP ボディ)

{"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

1 ユーザーの記憶数を返します。

mem.stats("alice")

実際のレスポンス

{"short_term_turns": 2, "total_memories": 4, "user_id": "alice"}

get

id で記憶を1件取得します - add や list_memories が返した id です。保存した原文とメタデータのみを返します。他のストアの id や、削除・無効化された記憶は 404 になります。

m = mem.get("alice", memory_id="576700aa-...")

response

{"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}

実際のレスポンス

{"memory": {"id": "8bd090de-...", "content": "the office moved to the seventh floor in June",
  "category": "general", "created_at": "2026-07-31T18:20:30.531518060+00:00", "event_date": null,
  "is_superseded": false, "superseded_by": null}, "user_id": "docs_livetest"}

list_memories

ストアに保存された記憶を一覧します。保存した原文とメタデータのみを返します。カーソルでページ送りします - 応答の next_cursor を次の呼び出しに渡します。

page = mem.list_memories("alice", limit=100)

response

{"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

カーソル管理なしで全ての記憶を巡回するか、ストア全体を一度に取得します。

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

削除

delete

id を指定して記憶を 1 件削除します。

mem.delete("alice", memory_id="576700aa-...")

実際のレスポンス

{"memory_id": "576700aa-f0e0-4c26-99a0-10e2d5b0d624", "status": "deleted"}

delete_all

1 ユーザーのすべてを消去します - 1 回の呼び出しで、GDPR 対応です。

mem.delete_all("alice")

実際のレスポンス

{"memories_deleted": 4, "status": "deleted", "user_id": "alice"}

エラーと信頼性

すべての失敗は型付きエラーです。個別(レート制限・認証・支払い)に捕捉するか、基底の 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

rate_limit

呼び出し直後に残り上限を読み、限界に達する前に速度を落とします。

mem.search("...", user_id="alice")
rl = mem.rate_limit   # {"limit": 150, "remaining": 3, "reset": ...}

search_self

自己記憶モデル（Scroll 1.2+）で 1 回の呼び出しから 2 つのレーンを受け取ります: 他者の発言とエージェント自身の発言を分けて返すので、読み手が話者を取り違えません。

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

選択中のモデルが実行できるエングラムとデリバリーフォームをサービスに問い合わせます。名前をハードコードすると、新しいエングラムが出た瞬間からそのコードには見えなくなります。

cat = mem.list_engrams()
[e["name"] for e in cat["engrams"]]   # ask, never hard-code

filters

検索をストアの一部に絞り込みます。ランキングの前に適用されるため、フィルタ内での最良の一致が返ります - 上位 N を後から絞ったものではありません。

mem.search("what did we decide", user_id="alice", filters={
    "categories": ["work"],
    "event_from": "2026-01-01",   # when it HAPPENED
})

キー: categories · event_from / event_to（内容がいつ起きたか - metadata.event_date）· time_from / time_to（いつ保存されたか）· min_importance。一覧にないキーは拒否ではなく破棄されるため、綴り間違いは黙って検索範囲を広げます。

idempotency_key

書き込みの再送を安全にします。再試行が自分側で起きるときに使います - 落ちて再実行されたジョブ、再配信するキュー。

mem.add("she prefers tea", "alice", idempotency_key=f"import:{row.id}")

キーは保存する対象から作ってください（import:row-42）。定数は使わないこと - 異なる 2 つの書き込みに同じキーを使うと最初の応答が再生され、2 つ目は黙って失われます。形式: [A-Za-z0-9._:-] の 1〜128 文字。

with_timeout / with_retries / with_deadline

すでに作ったクライアントに触れずに、呼び出し箇所だけを調整します。大きなバックフィルにはタイムアウトを延ばしたクローンを、自前の再試行ループの中では再試行を切ったクローンを使います。

timeout は 1 回の試行を区切ります。そのため再試行する呼び出しはそれより長く生き延びます — 既定では 1 回の呼び出しが 30 秒接続を保持し、待機し、再試行し、また試行します。deadline は代わりに呼び出し全体を区切ります。各試行は残り時間に切り詰められ、待機も予算を超えて眠りません。呼び出す側に実際の上限があるとき — 5 秒のリクエストハンドラのような場所で — 設定してください。

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

Product

Models

Start

Python - 全メソッドを 3 グループで。

モデルを選ぶ

list_models

ping

書き込み

add

add_turn

speaker

add_bulk

update

読み取り

search

recall

history

stats

get

list_memories

iter_memories · export_memories

削除

delete

delete_all

エラーと信頼性

rate_limit

search_self

list_engrams

filters

idempotency_key

with_timeout / with_retries / with_deadline