MemorySync
SDKs · Python

Python Memory Client

Build against the 35 public memory methods, their typed results, and explicit user scope. Use the API Reference when you need every field and status outcome.

Configure the client

client.py
import os
from memorysync import MemorySyncClient
client = MemorySyncClient(
api_key=os.environ["MEMORYSYNC_API_KEY"],
base_url="https://api.memorysync.io",
project_id=os.environ["MEMORYSYNC_PROJECT_ID"],
end_user_id="usr_7f3a9c2e",
)
ArgumentRequiredPurpose
api_keyYesServer-side memory API credential.
base_urlYesMemorySync HTTPS base URL.
project_idBy application scopeSelects the trusted project context.
end_user_idFor user-scoped memory callsStable opaque application-user identifier.
timeoutNoRequest timeout in seconds; defaults to 30.
transportNoInject an httpx transport when your application requires one.

All memory methods

MethodReturnsUse
add(text, ...)Memory | AddSkippedResponseStore one memory candidate.
bulk_add(items)BulkAddResponseSubmit 1–50 independent items and inspect each outcome.
query(query, ...)QueryResponseRetrieve up to the requested result window.
get(memory_id)MemoryRead one known record.
update(memory_id, ...)MemoryReplace supported descriptive fields.
forget(memory_ids, reason=None)list[int]Remove selected records and inspect confirmed IDs.
summarize(memory_ids, lossless=False)MemoryCreate a summary record from selected IDs.
compose(prompt_template, ...)ComposeResponseReceive a composed prompt and usage fields.
export_all()ExportResponseRead the authenticated principal export bundle.
create_import(file, ...)dictUpload a payload and start an import job.
get_import(job_id)dictPoll one job’s status and counters.
list_imports(...)dictRecent import jobs, newest first.
cancel_import(job_id)dictAsk the worker to stop between batches.
create_relation(from_memory_id, ...)RelationCreate a documented relationship between records.
add_turn(tenant_id=..., user_id=..., ...)dictStore one conversational turn verbatim (episodic ingestion).
recall(tenant_id=..., user_id=..., prompt=..., k=None)dictBuild a prompt-ready context block for an LLM call.
list_memories(tenant_id=..., user_id=..., ...)dictPage through a specific end user’s memories.
retrieve(query, ...)QueryResponseAlias of query against /memory/retrieve — identical semantics.
search_routed(query, route=None, ...)dictRoute a question to the best knowledge source and answer from it.
synthesize(query=None, memory_ids=None, ...)dictCompose an answer across several memories, with citations.
upload(file, filename=..., ...)AddResultIngest a document and store the memories extracted from its text.
batch_update(items)BatchUpdateResponseApply up to 100 metadata edits in one request.
refresh()dictRe-embed this end user’s memories; returns immediately (202).
status(memory_id)dictAsync ingestion status for one memory.
history(memory_id, limit=100, offset=0)HistoryResponseRecorded changes to one memory, oldest first.
feedback(memory_id, signal, comment=None)FeedbackResponseTell MemorySync whether a memory was useful; tunes ranking.
graph(limit=None, memory_id=None, depth=None)dictNodes and typed edges for this end user’s memory graph.
clusters(limit=None)dictSemantic clusters over this end user’s memories.
decisions(limit=None)dictContradictions and open decisions detected across memories.
resolve_decision(decision_id=None, ...)dictRecord which side of a contradiction wins.
intelligence(limit=None, scope=None, ...)dictThe intelligence report: themes, entities, patterns.
knowledge_stats()dictCounts and coverage for the knowledge base.
get_ontology()OntologyThe memory vocabulary in effect for this organization.
update_ontology(content_types=None, relation_types=None)OntologyReplace this organization’s additions to the vocabulary.
purge_user()always raisesGuard method: account erasure lives in the Control Plane SDK, never here.

Branch on Add outcomes

add_memory.py
from memorysync import AddSkippedResponse
result = client.add(
"The user prefers dark mode.",
source="settings",
metadata={"screen": "appearance"},
)
if isinstance(result, AddSkippedResponse):
print("not created", result.reason)
else:
print("created", result.id, result.text)

Reconcile every bulk item

bulk_add.py
from memorysync import BulkAddItem
response = client.bulk_add([
BulkAddItem(text="The launch is Friday.", tags=["plan"]),
BulkAddItem(text="The release needs legal approval.", tags=["constraint"]),
])
for item in response.results:
print(item.index, item.status, item.memory_ids, item.reason)

Retrieve and maintain records

retrieve_and_update.py
response = client.query(
"What constraints apply to the launch?",
k=5,
filters={"tags": ["constraint"]},
)
for memory in response.memories:
current = client.get(memory.id)
updated = client.update(current.id, tags=["constraint", "reviewed"])
print(updated.id, updated.tags)

Authorize destructive operations

forget.py
memory_ids = [101, 102]
deleted_ids = client.forget(memory_ids, reason="user_request")
if set(deleted_ids) != set(memory_ids):
print("Reconcile the IDs that were not confirmed")

Close the client

lifecycle.py
from memorysync import MemorySyncClient
with MemorySyncClient(
api_key="...",
base_url="https://api.memorysync.io",
project_id="...",
end_user_id="usr_7f3a9c2e",
) as client:
response = client.query("What should I remember?", k=3)

Reference and safety

Was this page helpful?