Memra Python SDK

Python client for the Memra memory API -- persistent, searchable, privacy-first memory for AI agents. EU-native, hosted in Helsinki.

Versioning: the SDK version tracks the Memra platform version. SDK 4.5.x targets Memra API v4.5.

Installation

pip install memra-sdk

Note: The package is installed as memra-sdk but imported as memra.

Quick Start (Sync)

from memra import MemraClient

client = MemraClient(api_key="memra_live_xxx")

# Store a memory
memory = client.memories.add(
    content="User is building a RAG pipeline for medical records",
    tenant_id="user_123",
    project_id="medical-assistant",
    type="fact",
    importance=9,
    tags=["project", "domain"],
)
print(memory.id)        # mem_abc123
print(memory.revision)  # read-your-writes token, e.g. 1042

# Recall memories by meaning -- wait_for_revision guarantees the write
# above is already indexed and searchable (read-your-writes)
results = client.memories.recall(
    query="What kind of product is this user building?",
    tenant_id="user_123",
    project_id="medical-assistant",
    limit=5,
    wait_for_revision=memory.revision,
)
for mem in results.data:
    print(f"[{mem.score:.3f}] {mem.content}")

client.close()

Quick Start (Async)

import asyncio
from memra import AsyncMemraClient

async def main():
    async with AsyncMemraClient(api_key="memra_live_xxx") as client:
        memory = await client.memories.add(
            content="User prefers async Python patterns",
            tenant_id="user_456",
            project_id="code-assistant",
        )

        results = await client.memories.recall(
            query="What Python patterns does this user prefer?",
            tenant_id="user_456",
            project_id="code-assistant",
        )

asyncio.run(main())

What's New in 4.5

Read-your-writes recall

Embeddings are generated asynchronously, so a memory written a moment ago may not be searchable yet. Every write response now carries a revision token; pass it to recall(wait_for_revision=...) and the server blocks until that write is indexed:

memory = client.memories.add(content="...", tenant_id="u1", project_id="p1")

results = client.memories.recall(
    query="...",
    tenant_id="u1",
    project_id="p1",
    wait_for_revision=memory.revision,  # deterministic write -> recall
)

Write responses also expose embedding_status ("pending" | "complete" | "failed").

Conflict detection on write

When a new memory contradicts existing knowledge, the create response tells you immediately via conflicts -- a list of MemoryConflict(memory_id, preview, confidence):

memory = client.memories.add(
    content="User switched from Postgres to SQLite",
    tenant_id="u1",
    project_id="p1",
)
for conflict in memory.conflicts or []:
    print(f"contradicts {conflict.memory_id} ({conflict.confidence:.2f}): {conflict.preview}")
    # one-call resolution:
    # client.memories.supersede(conflict.memory_id, memory.content)

Conflict detection is fail-open: it never blocks or fails the write.

Token-budget recall

Cap how much context recall may consume. Results are trimmed to fit max_tokens, and the response meta reports the budget accounting:

results = client.memories.recall(
    query="everything about this user's stack",
    tenant_id="u1",
    project_id="p1",
    max_tokens=800,
)
print(results.meta.token_budget)  # 800
print(results.meta.tokens_used)   # e.g. 763

Staleness signals on every recall item

Each recalled memory now carries staleness_score (0-100, 0 = fresh), staleness_status, and last_confirmed, so agents can decide whether to trust or re-verify a fact:

for mem in results.data:
    if mem.staleness_score > 50:
        print(f"stale ({mem.staleness_status}, last confirmed {mem.last_confirmed}): {mem.content}")

Feedback loop

Tell Memra which recalled memories were actually useful -- they get a scoring boost on future recalls:

result = client.memories.feedback(
    tenant_id="u1",
    project_id="p1",
    memory_ids=["mem_abc", "mem_def"],
)
print(result.updated)  # 2

Or skip the extra round trip and pass used_ids on the next recall:

client.memories.recall(
    query="...",
    tenant_id="u1",
    project_id="p1",
    used_ids=["mem_abc", "mem_def"],  # feedback from the previous recall
)

Entity graph

Memra's intelligence pipeline extracts entities from memories. Query the graph:

# Entities for a namespace, most-mentioned first
entities = client.entities.list(tenant_id="u1", project_id="p1")
for e in entities.entities:
    print(f"{e.name} ({e.type}) -- {e.memory_count} memories, pii={e.is_pii}")

# Filter by type, cap results
people = client.entities.list(
    tenant_id="u1", project_id="p1", entity_type="person", limit=20
)

# Memories mentioning an entity (metadata only -- fetch content via memories.get)
result = client.entities.memories("PostgreSQL", tenant_id="u1", project_id="p1")
print(result.entity, result.total)
for item in result.memories:
    full = client.memories.get(item.id)

PII entities appear under stable IDs, never raw values.

Recall Parameters

client.memories.recall(
    query="...",               # required: natural-language query
    tenant_id="u1",            # required: end-user / namespace ID
    project_id="p1",           # required: project ID
    limit=10,                  # max results
    type="fact",               # filter by memory type
    min_importance=5,          # minimum importance (1-10)
    scoring="default",         # scoring profile
    rerank=True,               # server-side reranking
    wait_for_revision=1042,    # block until this write revision is indexed
    max_tokens=800,            # token-budget recall (meta gains token_budget/tokens_used)
    not_tags=["archived"],     # exclude memories with any of these tags
    since="2026-01-01",        # only memories created on/after (ISO date)
    until="2026-06-30",        # only memories created on/before (ISO date)
    used_ids=["mem_abc"],      # feedback: useful IDs from the previous recall
)

API Coverage

Operation	Method	Description
client.memories.add()	POST /memories	Store a new memory
client.memories.list()	GET /memories	List memories with filters
client.memories.get(id)	GET /memories/:id	Get a single memory
client.memories.update(id)	PATCH /memories/:id	Update a memory
client.memories.delete(id)	DELETE /memories/:id	Delete a memory
client.memories.delete_tenant()	DELETE /memories	Bulk delete by tenant
client.memories.batch()	POST /memories/batch	Create up to 100 memories
client.memories.recall()	POST /memories/recall	Semantic search
client.memories.feedback()	POST /memories/feedback	Report useful memories (recall boost)
client.memories.supersede()	POST /memories/:id/supersede	Mark as superseded
client.memories.chain()	GET /memories/:id/chain	Get supersession chain
client.memories.promote()	POST /memories/:id/promote	Promote proposed → verified (returns PromotionResult)
client.memories.refresh()	POST /memories/:id/refresh	Reset staleness, return MemoryHealth
client.entities.list()	GET /entities	List entities in the namespace graph
client.entities.memories(name)	GET /entities/:name/memories	Memories mentioning an entity
client.projects.create()	POST /projects	Create a project
client.projects.list()	GET /projects	List projects
client.projects.get(id)	GET /projects/:id	Get a project
client.projects.delete(id)	DELETE /projects/:id	Delete a project
client.privacy.export()	GET /export	Data export (account-level)
client.privacy.namespace_export()	GET /namespaces/:id/data-export	Data export (per-tenant)
client.privacy.create_erasure_request()	POST /memories/:id/erasure-request	Request erasure
client.privacy.get_erasure_request()	GET /memories/:id/erasure-request	Check erasure status
client.usage.get()	GET /usage	Get account usage

Privacy & Data Protection

Memra is privacy-first. The Python SDK provides access to data export and erasure endpoints.

Data Export

# Export all account data
data = client.privacy.export()
print(data.exported_at)

# Export namespace data (per-tenant)
data = client.privacy.namespace_export("tenant_123")

# Export namespace data filtered by project
data = client.privacy.namespace_export("tenant_123", project_id="proj_1")

Data Erasure

# Request erasure of a memory
request = client.privacy.create_erasure_request("mem_abc123")
print(request.status)  # 'pending'

# Check erasure status
status = client.privacy.get_erasure_request("mem_abc123")
print(status.status)  # 'completed'

Erasure is thorough: flat files, database index rows, Redis cache entries, and audit log entries are all purged.

Error Handling

All API errors are mapped to typed exceptions:

from memra import MemraClient
from memra.exceptions import (
    MemraError,          # Base class for all errors
    MemraAuthError,      # 401 Unauthorized
    MemraNotFoundError,  # 404 Not Found
    MemraValidationError,# 422 Unprocessable Entity
    MemraQuotaError,     # 429 Rate Limited
    MemraServerError,    # 5xx Server Error
)

client = MemraClient(api_key="memra_live_xxx")

try:
    memory = client.memories.get("mem_nonexistent")
except MemraNotFoundError as e:
    print(f"Not found: {e} (status={e.status_code})")
except MemraAuthError:
    print("Invalid API key")
except MemraError as e:
    print(f"API error: {e}")

Configuration

# Default: Memra cloud
client = MemraClient(api_key="memra_live_xxx")

# Self-hosted instance
client = MemraClient(
    api_key="memra_live_xxx",
    base_url="https://yourdomain.com/api/v1",
)

# Custom timeout (default: 10 seconds)
client = MemraClient(
    api_key="memra_live_xxx",
    timeout=30.0,
)

Requirements

• Python 3.9+
• httpx >= 0.27
• pydantic >= 2.0

License

MIT