Persistent memory for AI agents. Not vector search — statistical learning backed by Hebbian associations.
pip install mnemoversefrom mnemoverse import MnemoClient
client = MnemoClient(api_key="mk_live_YOUR_KEY")
# Store a memory
result = client.write(
"Retry with exponential backoff fixed the timeout issue",
concepts=["retry", "backoff", "timeout"]
)
# Query — Hebbian associations expand "timeout" → "retry", "backoff"
memories = client.read("how to handle timeouts?")
# Report outcome — the system learns what works
client.feedback(
atom_ids=[item.atom_id for item in memories.items],
outcome=1.0,
query_concepts=memories.query_concepts
)from mnemoverse import AsyncMnemoClient
async with AsyncMnemoClient(api_key="mk_live_YOUR_KEY") as client:
result = await client.write("async memory", concepts=["async"])
memories = await client.read("what about async?")- Circuit breaker — 5 failures → open → 30s half-open → probe
- Retry with backoff — 3 attempts, rate-limit-aware
- Sync + async —
MnemoClientfor scripts,AsyncMnemoClientfor FastAPI - Type-safe — Pydantic models, full type hints
| Method | Description |
|---|---|
write(content, concepts, domain, metadata) |
Store a memory |
write_batch(items) |
Store up to 500 memories |
read(query, top_k, domain, since, until, order_by, exclude_author) |
Semantic search — "what do I know about X" |
recent(domain, since, until, exclude_author, limit, cursor) |
Newest-first feed — "what happened lately" |
feedback(atom_ids, outcome) |
Report success/failure |
stats() |
Memory statistics |
health() |
API health check |
Every method exists on both MnemoClient (sync) and AsyncMnemoClient (async).
read() answers what do I know about X and ranks by relevance. recent()
answers what happened lately and is complete within one scope by
construction — nothing is skipped, which a semantic search cannot promise.
Reach for recent() to resume after a break or to catch up on a shared room.
from mnemoverse import MnemoClient
client = MnemoClient(api_key="mk_live_...")
# Catch up on a shared room. Rooms are SEPARATE stores: pass the address as
# `domain`, or an unscoped feed will not cover them.
page = client.recent(domain="xroom:room_01ABC", since="2026-08-01T00:00:00Z", limit=20)
for item in page.items:
print(item.created_at, item.content)
if page.next_cursor:
page = client.recent(domain="xroom:room_01ABC", cursor=page.next_cursor)Read items carry created_at and provenance (who wrote it, where from).
MIT