Unified memory gateway for AI agents. One interface, multiple memory providers.
memio lets you swap between memory backends (Mem0, Zep, Chroma, Letta, Qdrant, Supermemory) without changing your application code. Define what memory capabilities you need — facts, conversation history, documents, knowledge graphs — and plug in any supported provider.
- Protocol-based architecture — providers implement Python protocols, so you can mix and match or bring your own
- Async-first — all operations are
async/await - Zero production dependencies — install only the providers you need
- Composable — use Mem0 for facts, Zep for history, and Chroma for documents in the same client
- Multi-tenant — scope data by
user_idoragent_id - Consistent error handling — all provider errors wrapped in
ProviderErrorwith context
pip install memio
Install with providers:
pip install memio[mem0] # Mem0 provider pip install memio[zep] # Zep provider pip install memio[chroma] # Chroma provider pip install memio[letta] # Letta provider pip install memio[qdrant] # Qdrant provider pip install memio[supermemory] # Supermemory provider pip install memio[all] # All providers pip install memio[server] # REST API server
from memio import Memio from memio.providers.mem0 import Mem0FactAdapter from memio.providers.zep import ZepHistoryAdapter from memio.providers.chroma import ChromaDocumentAdapter import chromadb client = Memio( facts=Mem0FactAdapter(api_key="your-mem0-key"), history=ZepHistoryAdapter(api_key="your-zep-key"), documents=ChromaDocumentAdapter( client=chromadb.EphemeralClient(), collection_name="my-docs", ), ) # Store and retrieve facts fact = await client.facts.add(content="likes coffee", user_id="alice") results = await client.facts.search(query="coffee", user_id="alice") # Manage conversation history from memio import Message await client.history.add( session_id="session-1", user_id="alice", messages=[ Message(role="user", content="hello"), Message(role="assistant", content="hi there"), ], ) messages = await client.history.get(session_id="session-1") # Store and search documents doc = await client.documents.add(content="memio is a memory gateway") results = await client.documents.search(query="memory")
memio defines four memory store protocols. Each provider implements one or more:
| Store | Purpose | Mem0 | Zep | Chroma | Letta | Qdrant | Supermemory |
|---|---|---|---|---|---|---|---|
FactStore |
Structured facts about users/agents | yes | yes | - | yes | - | yes |
HistoryStore |
Conversation message history | - | yes | - | yes | - | - |
DocumentStore |
Document storage with semantic search | - | - | yes | yes | yes | yes |
GraphStore |
Knowledge graph triples | yes | yes | - | - | - | - |
fact = await store.add(content="prefers dark mode", user_id="alice") fact = await store.get(fact_id=fact.id) results = await store.search(query="preferences", user_id="alice") updated = await store.update(fact_id=fact.id, content="prefers light mode") all_facts = await store.get_all(user_id="alice") await store.delete(fact_id=fact.id) await store.delete_all(user_id="alice")
await store.add(session_id="s1", user_id="alice", messages=[Message(role="user", content="hello")]) messages = await store.get(session_id="s1", limit=50) results = await store.search(session_id="s1", query="hello") sessions = await store.get_all(user_id="alice") await store.delete(session_id="s1") await store.delete_all(user_id="alice")
doc = await store.add(content="some text", metadata={"source": "wiki"}) doc = await store.get(doc_id=doc.id) results = await store.search(query="text", limit=10) all_docs = await store.get_all(limit=100) updated = await store.update(doc_id=doc.id, content="updated text") await store.delete(doc_id=doc.id) await store.delete_all()
from memio import Triple await store.add( triples=[Triple(subject="Alice", predicate="likes", object="coffee")], user_id="alice", ) result = await store.get(entity="Alice", user_id="alice") result = await store.search(query="coffee", user_id="alice") await store.delete_all(user_id="alice")
from memio import Fact, Message, Document, Triple, GraphResult # Fact — a stored piece of knowledge Fact(id, content, user_id, agent_id, metadata, score, created_at, updated_at) # Message — a conversation message Message(role, content, metadata, timestamp, name) # Document — a stored document Document(id, content, metadata, score, created_at, updated_at) # Triple — a knowledge graph triple Triple(subject, predicate, object, metadata) # GraphResult — result from graph queries GraphResult(triples, nodes, scores)
Use memio from any language (JavaScript, TypeScript, Go, etc.) over HTTP:
pip install memio[server,mem0,chroma]
Create memio-server.yaml:
stores: facts: provider: mem0 config: api_key: "${MEM0_API_KEY}" documents: provider: chroma config: collection_name: "my-docs"
Start the server:
memio-server
# Open http://localhost:8080/docs for interactive API docsUse from JavaScript:
// Add a fact const fact = await fetch("http://localhost:8080/v1/facts", { method: "POST", headers: { "Content-Type": "application/json" }, body: JSON.stringify({ content: "likes coffee", user_id: "alice" }), }).then(r => r.json()); // Search const results = await fetch("http://localhost:8080/v1/facts/search", { method: "POST", headers: { "Content-Type": "application/json" }, body: JSON.stringify({ query: "preferences", user_id: "alice" }), }).then(r => r.json());
See the REST API guide for full endpoint reference, authentication, Docker deployment, and configuration options.
Implement any protocol to create your own provider:
from memio import Memio, DocumentStore, Document class MyDocumentStore: async def add(self, *, content, doc_id=None, metadata=None): # your implementation return Document(id="...", content=content, metadata=metadata) async def get(self, *, doc_id): ... async def search(self, *, query, limit=10, filters=None): ... async def update(self, *, doc_id, content, metadata=None): ... async def delete(self, *, doc_id): ... async def get_all(self, *, limit=100, filters=None): ... async def delete_all(self): ... # memio validates the protocol at runtime client = Memio(documents=MyDocumentStore())
All provider errors are wrapped in ProviderError:
from memio import ProviderError, NotSupportedError, MemioError try: await client.facts.delete(fact_id="123") except NotSupportedError as e: print(e.hint) # "use delete_all" except ProviderError as e: print(e.provider) # "mem0", "zep", etc. print(e.operation) # "search", "add", etc. print(e.cause) # original exception except MemioError: # base class for all memio errors ...
NotSupportedError is a subclass of ProviderError, raised when a provider doesn't support a specific operation (e.g., individual delete on Zep). It includes an optional hint suggesting what to do instead. Unlike other ProviderError subclasses, its cause is a NotImplementedError (not an underlying SDK exception).
Mem0 — content may be rephrased by Mem0's LLM. Duplicate content is deduplicated automatically.
Zep — graph operations are eventually consistent. graph.add sends text to an LLM for asynchronous fact extraction. Individual fact/triple deletion is not supported; use delete_all instead.
Chroma — uses a local client you provide. No API key required for ephemeral or persistent local usage.
Letta — supports both Letta Cloud (api_key) and self-hosted (base_url). All stores are agent-scoped via agent_id. Facts and documents both map to archival passages. No graph support.
from memio import Memio from memio.providers.letta import LettaFactAdapter, LettaHistoryAdapter, LettaDocumentAdapter client = Memio( facts=LettaFactAdapter(api_key="letta-xxx", agent_id="agent-123"), history=LettaHistoryAdapter(api_key="letta-xxx", agent_id="agent-123"), documents=LettaDocumentAdapter(api_key="letta-xxx", agent_id="agent-123"), )
Qdrant — uses qdrant-client with built-in fastembed for automatic embeddings. Supports in-memory, on-disk, and Qdrant Cloud deployments. No API key required for local usage.
Supermemory — cloud-hosted memory API. Content is auto-processed by an LLM for fact extraction. get() and delete_all() on FactStore are not supported. Documents are processed asynchronously after add().
git clone https://github.com/y3zai/memio.git cd memio python -m venv .venv && source .venv/bin/activate pip install -e ".[all,dev]" # Run unit tests pytest # Put OPENAI_API_KEY, MEM0_API_KEY, ZEP_API_KEY, and SUPERMEMORY_API_KEY in .env, then run integration tests pytest -m integration -v
Pytest loads .env automatically from the repository root for local test runs.
MIT