MemorySync
SDKs · Python

Python Async Usage

Use AsyncMemorySyncClient when the surrounding application already uses asyncio. It mirrors all 31 memory methods and provides native asynchronous cleanup.

When async is the right choice

Python call styles

Use the sync client

Best for ordinary scripts and synchronous request paths.

  • Direct method calls.
  • with context manager.
  • close() cleanup.

Use the async client

Best when the caller already runs an event loop.

  • Await every operation.
  • async with context manager.
  • aclose() cleanup.

Complete async example

async_quickstart.py
import asyncio
import os
from memorysync import AsyncMemorySyncClient
async def main() -> None:
async with AsyncMemorySyncClient(
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",
) as client:
await client.add("The user prefers weekly progress summaries.")
response = await client.query("How often should I send updates?", k=3)
for memory in response.memories:
print(memory.id, memory.text)
asyncio.run(main())

The same 35 operations

SynchronousAsynchronous
client.add(...)await client.add(...)
client.bulk_add(...)await client.bulk_add(...)
client.query(...)await client.query(...)
client.get(...)await client.get(...)
client.update(...)await client.update(...)
client.forget(...)await client.forget(...)
client.summarize(...)await client.summarize(...)
client.compose(...)await client.compose(...)
client.export_all()await client.export_all()
client.create_import(...)await client.create_import(...)
client.get_import(...)await client.get_import(...)
client.list_imports(...)await client.list_imports(...)
client.cancel_import(...)await client.cancel_import(...)
client.create_relation(...)await client.create_relation(...)
client.add_turn(...)await client.add_turn(...)
client.recall(...)await client.recall(...)
client.list_memories(...)await client.list_memories(...)
client.retrieve(...)await client.retrieve(...)
client.search_routed(...)await client.search_routed(...)
client.synthesize(...)await client.synthesize(...)
client.upload(...)await client.upload(...)
client.batch_update(...)await client.batch_update(...)
client.refresh()await client.refresh()
client.status(...)await client.status(...)
client.history(...)await client.history(...)
client.feedback(...)await client.feedback(...)
client.graph(...)await client.graph(...)
client.clusters(...)await client.clusters(...)
client.decisions(...)await client.decisions(...)
client.resolve_decision(...)await client.resolve_decision(...)
client.intelligence(...)await client.intelligence(...)
client.knowledge_stats()await client.knowledge_stats()
client.get_ontology()await client.get_ontology()
client.update_ontology(...)await client.update_ontology(...)

Bound application concurrency

bounded_queries.py
import asyncio
from memorysync import AsyncMemorySyncClient, QueryResponse
async def query_many(
client: AsyncMemorySyncClient,
questions: list[str],
*,
concurrency: int = 4,
) -> list[QueryResponse]:
semaphore = asyncio.Semaphore(concurrency)
async def run(question: str) -> QueryResponse:
async with semaphore:
return await client.query(question, k=3)
return await asyncio.gather(*(run(question) for question in questions))

Own the client lifecycle

Async client lifecycle
  1. 01

    Create at a clear application boundary

    START

    Construct the client where your service manages shared resources.

  2. 02

    Reuse inside that boundary

    USE

    Pass the client to code that needs memory operations.

  3. 03

    Close during shutdown

    STOP

    Prefer async with or call await client.aclose().

Handle failures with the same error classes

Async methods raise the same six public error classes as synchronous methods. Catch permanent errors before the base MemorySyncError, and do not assume an interrupted write failed.

Was this page helpful?