# API Reference The three public client classes: `Memory`, `AsyncMemory`, `UserHandle`. ## `Memory` The synchronous front door. Long-lived per process. Storage resolves lazily on first call — no I/O in `__init__`. ### Constructor ```python Memory( *, api_key: str | None = None, base_url: str | None = None, path: str | os.PathLike[str] | None = None, validate_user_id: Callable[[str], None] | None = None, timeout: float | httpx.Timeout = 10.0, max_retries: int = 2, suppress_async_warning: bool | None = None, transport: httpx.BaseTransport | None = None, ) ``` Every argument is keyword-only. - `api_key` — reserved for v0.2 remote backend; defaults to `ATLASO_API_KEY` env. - `base_url` — reserved for v0.2; defaults to `ATLASO_BASE_URL` env or `https://api.atlaso.dev`. - `path` — SQLite location. If `None`, resolved via the 5-step path walker (see [Configuration](./configuration.md)). - `validate_user_id` — optional callback called inside `for_user()` and every lower-level call; raise to block. - `timeout` — httpx timeout for the future remote backend. - `max_retries` — transport-level retries (network + 5xx + 429). - `suppress_async_warning` — silence `SyncInAsyncWarning`. Lever order: this arg > `ATLASO_ASYNC_WARNINGS=0` > `warnings.filterwarnings(...)`. - `transport` — inject an `httpx.BaseTransport` for tests. **Passing an `httpx.Client` raises `ConfigValidationError`** — the threat model is silent cross-tenant Authorization-header leak. Use `httpx.MockTransport(...)` for tests. ### `for_user(user_id) -> UserHandle` Returns a frozen handle bound to `user_id`. Pre-fills `user_id` on every call so identity can't be fumbled per-method. ### `add(text, *, user_id, polarity="open", evidence_grade="anecdotal", scope=None, tags=None, contradicts=None, author=None) -> AddResult` Write a deposit. Raises `InputValidationError`, `DepositRejectedError`, or `MissingContradictsError`. ### `recall(query, *, user_id, limit=10, scope=None) -> SearchResults` Dispersion-aware search. `limit` is bounded `1..1000`. The returned `SearchResults` has bag-level `has_disagreement` / `is_confident` flags computed across **all** bags before slicing. ### `get(deposit_id, *, user_id) -> Deposit | None` Fetch one deposit by id. **Returns `None` on miss** — deliberately distinct from `NotFoundError`, for dict.get muscle memory. ### `peek(user_id, *, limit=10) -> PeekView` REPL/Jupyter snapshot — recent deposits + FMI + a recent-disagreement flag. ### `list_recent(*, user_id, limit=20, offset=0) -> list[Deposit]` Newest-first paginated list. ### `contradict(new_text, contradicts, *, reason, user_id) -> AddResult` Atomic supersede. Writes a new deposit and marks one or more existing deposits as superseded. `reason` is required; empty `contradicts` list raises `InputValidationError`. ### `retract(deposit_id, *, reason, user_id, hard_delete=False, force=False) -> RetractResult` Soft retract by default (the deposit becomes invisible to recall but survives in the file). `hard_delete=True` is irreversible. `force` is reserved for v0.2 server-side protected-deposit flags; no effect on the local backend today. ### `health(*, user_id, window_days=30) -> Diagnostics` FMI + four pillars. `window_days` bounded `1..365`. ### `add_many(items, *, user_id, on_gate_reject="skip") -> AddManyResult` Bulk import. Each `AddItem` requires an `idempotency_key`. Skip-and-report semantics — successful items in `result.committed`, idempotent replays in `result.duplicates`, gate rejections in `result.failed`. ### `close()` / context manager ```python with Memory() as m: user = m.for_user("alice") user.add("…") # backend closed, httpx client closed ``` A caller-supplied transport is **not** closed for you. ### `update()` is a typo-blocker ```python m.update("anything") # AttributeError: deposits are immutable. # Use m.contradict(new_text, contradicts=[old.id], reason="…") instead. ``` --- ## `AsyncMemory` The async mirror of `Memory`. Same kwargs and method shapes; data-plane methods are awaitable. ### Differences from `Memory` - `transport` must be `httpx.AsyncBaseTransport`. Passing `httpx.AsyncClient` raises `ConfigValidationError`. - `aclose()` instead of `close()`. - `__aenter__` / `__aexit__` instead of sync context manager. - `for_user(user_id)` stays **sync** — returns an `AsyncUserHandle` with no I/O. Only the handle's data methods are awaitable. - `update()` is sync (it just raises with a pointer at `contradict()`). - `suppress_async_warning` accepted for parity but is a no-op. - Calling `AsyncMemory` from sync code raises immediately rather than warning. ### Canonical FastAPI shape ```python from contextlib import asynccontextmanager from fastapi import FastAPI, Depends from atlaso import AsyncMemory memory: AsyncMemory | None = None @asynccontextmanager async def lifespan(app: FastAPI): global memory memory = AsyncMemory() yield await memory.aclose() app = FastAPI(lifespan=lifespan) @app.post("/remember") async def remember(text: str, user_id: str = Depends(current_user)): user = memory.for_user(user_id) # sync, no I/O result = await user.add(text) return {"id": result.id} @app.get("/recall") async def recall(q: str, user_id: str = Depends(current_user)): user = memory.for_user(user_id) hits = await user.recall(q) return { "verdict": hits.explain(), "is_confident": hits.is_confident, "has_disagreement": hits.has_disagreement, "items": [ { "content": h.content, "is_confident": h.is_confident, "has_disagreement": h.has_disagreement, "agreement_score": h.agreement_score, } for h in hits ], } ``` > v0.1 wraps the sync SQLite backend in `asyncio.to_thread()`. A native async backend ships in v0.2. The public API stays the same. --- ## `UserHandle` Frozen slotted dataclass returned by `Memory.for_user(...)`. Holds the client and an authenticated `user_id`. Every method pre-fills `user_id` and delegates. ```python m = Memory(validate_user_id=allowlist_check) user = m.for_user(auth.user.id) # validate runs once here user.add("…") hits = user.recall("…") ``` Method surface mirrors `Memory` minus the `user_id=` kwarg: - `user.add(text, *, polarity="open", evidence_grade="anecdotal", scope=None, tags=None, contradicts=None, author=None)` - `user.recall(query, *, limit=10, scope=None)` - `user.get(deposit_id)` - `user.peek(*, limit=10)` - `user.list_recent(*, limit=20, offset=0)` - `user.contradict(new_text, contradicts, *, reason)` - `user.retract(deposit_id, *, reason, hard_delete=False, force=False)` - `user.health(*, window_days=30)` - `user.add_many(items, *, on_gate_reject="skip")` `user.user_id` exposes the bound id. The handle is frozen — to rebind, call `for_user(...)` again on the underlying `Memory`. There is no `handle.rebind()`. `AsyncMemory.for_user(...)` returns an `AsyncUserHandle` with the same method names but every data call awaitable. --- **Source:** **Edit on GitHub:** **Updated:** 2026-05-12