Skip to content

Python SDK ​

Installation ​

bash
pip install mnemoverse

Requires Python 3.10+.

Quick Start ​

python
from 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"],
    domain="engineering"
)
print(f"Stored: {result.stored}, ID: {result.atom_id}")

# Query memories
memories = client.read("how to handle timeouts?", top_k=5)
for item in memories.items:
    print(f"[{item.relevance:.2f}] {item.content}")

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

Client Configuration ​

python
client = MnemoClient(
    api_key="mk_live_YOUR_KEY",
    base_url="https://core.mnemoverse.com",  # default
    timeout=10.0,                            # seconds, default: 10
    max_retries=3,                           # default: 3
)

Methods ​

write() ​

Store a single memory.

python
result = client.write(
    content="Caching reduced API latency by 40%",
    concepts=["caching", "latency", "optimization"],
    domain="engineering",
    metadata={"source": "incident-report", "date": "2026-04-08"},
    external_ref="incident-42"  # idempotent — won't duplicate
)
# result.stored: bool
# result.atom_id: UUID | None
# result.importance: float
# result.reason: str

write_batch() ​

Store up to 500 memories in one call.

python
items = [
    {"content": "Memory 1", "concepts": ["a"]},
    {"content": "Memory 2", "concepts": ["b"]},
]
result = client.write_batch(items)
# result.total_count: int
# result.stored_count: int
# result.results: list[WriteBatchItemResult]

read() ​

Query memories with semantic search + Hebbian expansion.

python
memories = client.read(
    query="how to optimize database queries?",
    top_k=10,
    domain="engineering",
    min_relevance=0.3,
    include_associations=True
)
# memories.items: list[MemoryItem]
# memories.episodic_hit: bool
# memories.query_concepts: list[str]
# memories.expanded_concepts: list[str]
# memories.search_time_ms: float

feedback() ​

Report outcomes to update valence and Hebbian associations.

python
response = client.feedback(
    atom_ids=[memories.items[0].atom_id],
    outcome=0.8,  # -1.0 (failure) to +1.0 (success)
    query_concepts=memories.query_concepts  # enables co-activation learning
)
# response.updated_count: int
# response.avg_valence: float
# response.coactivation_edges: int

stats() ​

Get memory statistics.

python
stats = client.stats()
# stats.total_atoms: int
# stats.hebbian_edges: int
# stats.avg_valence: float
# stats.domains: list[str]

health() ​

Check API health.

python
health = client.health()
# health.status: str ("ok")
# health.database: bool

Async Client ​

For async applications (FastAPI, Discord bots, etc.):

python
from mnemoverse import AsyncMnemoClient

client = AsyncMnemoClient(api_key="mk_live_YOUR_KEY")

result = await client.write("async memory", concepts=["async"])
memories = await client.read("what about async?")
await client.feedback(atom_ids=[...], outcome=1.0)

Error Handling ​

python
from mnemoverse import MnemoClient, MnemoAuthError, MnemoRateLimitError, MnemoError

client = MnemoClient(api_key="mk_live_YOUR_KEY")

try:
    result = client.read("query")
except MnemoAuthError:
    print("Invalid API key")
except MnemoRateLimitError as e:
    print(f"Rate limited. Retry after {e.retry_after}s")
except MnemoError as e:
    print(f"API error: {e.message}")

Integration Examples ​

LangChain Tool ​

python
from langchain.tools import tool
from mnemoverse import MnemoClient

memory = MnemoClient(api_key="mk_live_YOUR_KEY")

@tool
def remember(content: str, concepts: list[str]) -> str:
    """Store a memory for future reference."""
    result = memory.write(content, concepts=concepts)
    return f"Stored: {result.stored}, importance: {result.importance:.2f}"

@tool
def recall(query: str) -> str:
    """Search memories for relevant information."""
    results = memory.read(query, top_k=5)
    return "\n".join(f"- {item.content}" for item in results.items)

CrewAI Agent ​

python
from crewai import Agent, Task
from mnemoverse import MnemoClient

memory = MnemoClient(api_key="mk_live_YOUR_KEY")

# Store experience after task completion
def on_task_complete(task_output):
    memory.write(
        content=f"Task result: {task_output.summary}",
        concepts=task_output.tags,
        domain="crewai"
    )
    memory.feedback(atom_ids=[...], outcome=task_output.score)

Source Code ​

The SDK is open source: github.com/mnemoverse/mnemoverse-sdk-python