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.get("MEMORYSYNC_PROJECT_ID") or None,
    end_user_id="usr_7f3a9c2e",
)

Argument	Required	Purpose
`api_key`	Yes	Server-side memory API credential.
`base_url`	Yes	MemorySync HTTPS base URL.
`project_id`	By application scope	Selects the trusted project context.
`end_user_id`	For user-scoped memory calls	Stable opaque application-user identifier.
`timeout`	No	Request timeout in seconds; defaults to 30.
`transport`	No	Inject an `httpx` transport when your application requires one.

All memory methods

Method	Returns	Use
`add(text, ...)`	`Memory \| AddSkippedResponse`	Store one memory candidate.
`bulk_add(items)`	`BulkAddResponse`	Submit 1–50 independent items and inspect each outcome.
`query(query, ...)`	`QueryResponse`	Retrieve up to the requested result window.
`get(memory_id)`	`Memory`	Read one known record.
`update(memory_id, ...)`	`Memory`	Replace supported descriptive fields.
`forget(memory_ids, reason=None)`	`list[int]`	Remove selected records and inspect confirmed IDs.
`summarize(memory_ids, lossless=False)`	`Memory`	Create a summary record from selected IDs.
`compose(prompt_template, ...)`	`ComposeResponse`	Receive a composed prompt and usage fields.
`export_all()`	`ExportResponse`	Read the authenticated principal export bundle.
`create_import(file, ...)`	`dict`	Upload a payload and start an import job.
`get_import(job_id)`	`dict`	Poll one job’s status and counters.
`list_imports(...)`	`dict`	Recent import jobs, newest first.
`cancel_import(job_id)`	`dict`	Ask the worker to stop between batches.
`create_relation(from_memory_id, ...)`	`Relation`	Create a documented relationship between records.
`add_turn(tenant_id=..., user_id=..., text=..., role=None, ...)`	`dict`	Send one conversation turn to fact extraction. Only a user turn is extracted; the durable facts in it become memories and the turn text is not stored (`memory_id` is `None`).
`append_history(tenant_id=..., user_id=..., session_id=..., turns=..., ...)`	`AppendHistoryResponse`	Append 1–100 turns to a session transcript your agent reads back; user turns are also sent to fact extraction unless `extract=False`.
`list_history(tenant_id=..., user_id=..., session_id=..., ...)`	`ListHistoryResponse`	A session’s turns, oldest first.
`delete_history(tenant_id=..., user_id=..., session_id=None, ids=None, ...)`	`DeleteHistoryResponse`	Delete a session or specific turns; `delete_derived_facts=True` also deletes the facts extracted from them.
`put_state(tenant_id=..., user_id=..., namespace=..., key=..., value=..., ...)`	`PutStateResponse`	Store a key’s value in framework state, replacing the previous one.
`get_state(tenant_id=..., user_id=..., namespace=..., key=...)`	`GetStateResponse`	The current value of one key.
`list_state(tenant_id=..., user_id=..., namespace_prefix=None, ...)`	`ListStateResponse`	Current values under a namespace prefix, newest first.
`search_state(tenant_id=..., user_id=..., query=..., ...)`	`SearchStateResponse`	Rank current values against a query.
`delete_state(tenant_id=..., user_id=..., namespace=..., keys=None, ...)`	`DeleteStateResponse`	Delete keys or a whole namespace.
`list_state_namespaces(tenant_id=..., user_id=..., ...)`	`ListStateNamespacesResponse`	Namespaces holding at least one value for this user.
`recall(tenant_id=..., user_id=..., prompt=..., k=None)`	`dict`	Build a prompt-ready context block for an LLM call.
`list_memories(tenant_id=..., user_id=..., ...)`	`dict`	Page through a specific end user’s memories.
`retrieve(query, ...)`	`QueryResponse`	Alias of `query` against `/memory/retrieve` — identical semantics.
`search_routed(query, route=None, ...)`	`dict`	Route a question to the best knowledge source and answer from it.
`synthesize(query=None, memory_ids=None, ...)`	`dict`	Compose an answer across several memories, with citations.
`upload(file, filename=..., ...)`	`AddResult`	Ingest a document and store the memories extracted from its text.
`batch_update(items)`	`BatchUpdateResponse`	Apply up to 100 metadata edits in one request.
`refresh()`	`dict`	Re-embed this end user’s memories; returns immediately (202).
`status(memory_id)`	`dict`	Async ingestion status for one memory.
`history(memory_id, limit=100, offset=0)`	`HistoryResponse`	Recorded changes to one memory, oldest first.
`feedback(memory_id, signal, comment=None)`	`FeedbackResponse`	Tell MemorySync whether a memory was useful; tunes ranking.
`graph(limit=None, memory_id=None, depth=None)`	`dict`	Nodes and typed edges for this end user’s memory graph.
`clusters(limit=None)`	`dict`	Semantic clusters over this end user’s memories.
`decisions(limit=None)`	`dict`	Contradictions and open decisions detected across memories.
`resolve_decision(decision_id=None, ...)`	`dict`	Record which side of a contradiction wins.
`intelligence(limit=None, scope=None, ...)`	`dict`	The intelligence report: themes, entities, patterns.
`knowledge_stats()`	`dict`	Counts and coverage for the knowledge base.
`get_ontology()`	`Ontology`	The memory vocabulary in effect for this organization.
`update_ontology(content_types=None, relation_types=None)`	`Ontology`	Replace this organization’s additions to the vocabulary.
`purge_user()`	always raises	Guard 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

Memory API Reference

See request fields and successful response variants.

Types & Errors

Use dataclasses and typed exception handling.

Memory Contract

Understand customer-visible record fields.

Retrieval Contract

Consume candidate results safely.

Was this page helpful?

← Previous

Quickstart

Async Usage