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
Use the sync client
Best for ordinary scripts and synchronous request paths.
- Direct method calls.
withcontext manager.close()cleanup.
Use the async client
Best when the caller already runs an event loop.
- Await every operation.
async withcontext manager.aclose()cleanup.
Complete async example
async_quickstart.py
import asyncioimport osfrom memorysync import AsyncMemorySyncClientasync 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
| Synchronous | Asynchronous |
|---|---|
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 asynciofrom memorysync import AsyncMemorySyncClient, QueryResponseasync 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
- 01
Create at a clear application boundary
STARTConstruct the client where your service manages shared resources.
- 02
Reuse inside that boundary
USEPass the client to code that needs memory operations.
- 03
Close during shutdown
STOPPrefer
async withor callawait 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?