From 913dad029fa3cee92227764b640bc9a690c62a6b Mon Sep 17 00:00:00 2001 From: Shah Hussain <95882307+shhahhussain@users.noreply.github.com> Date: Wed, 14 Oct 2026 10:30:51 +0500 Subject: [PATCH 1/2] feat!: rewrite to the current Mnemo API contract (0.2.0) The published 0.1.2 predates the container-scoped API and every request it sends is rejected: POST /v1/memories got {content} instead of {containerTag, items:[...]}, POST /v1/search got {query} instead of {q}, and responses were parsed as hits/workspaceId which no longer exist. - Containers are now first-class: default_container_tag on the client (env GETMNEMO_CONTAINER_TAG) or container_tag per call, format : - add() posts {containerTag, items:[{content, metadata?}]} and returns the typed AddResponse (items, stats, receipt); new add_many() for 1-100 items - search() posts {q, containerTag, limit?} and returns list[SearchHit] with the real response fields (resultType, score, scopeKey, polarity, ...) - search() treats 404 'Container not found' as empty results and tolerates the upcoming empty-array response shape - workspace_id is deprecated-optional: identity comes from the API key; the x-workspace-id header is only sent when explicitly configured - actor_id removed everywhere (no longer part of the API) - update/delete/list thread the container as query params - canonical import is now 'getmnemo' (matches the install name and the LangChain adapter); 'mnemo' remains as a back-compat alias module - tests pin the wire shapes: bodies carry ONLY API-accepted keys (unknown-key regression), 404->[] on search, env fallbacks, both clients --- README.md | 58 ++- pyproject.toml | 8 +- src/getmnemo/__init__.py | 54 +++ src/getmnemo/client.py | 562 ++++++++++++++++++++++++++++++ src/{mnemo => getmnemo}/errors.py | 0 src/getmnemo/models.py | 92 +++++ src/getmnemo/py.typed | 0 src/mnemo/__init__.py | 54 ++- src/mnemo/client.py | 402 --------------------- src/mnemo/models.py | 46 --- tests/test_client.py | 373 ++++++++++++++++---- 11 files changed, 1077 insertions(+), 572 deletions(-) create mode 100644 src/getmnemo/__init__.py create mode 100644 src/getmnemo/client.py rename src/{mnemo => getmnemo}/errors.py (100%) create mode 100644 src/getmnemo/models.py create mode 100644 src/getmnemo/py.typed delete mode 100644 src/mnemo/client.py delete mode 100644 src/mnemo/models.py diff --git a/README.md b/README.md index 25f8df1..6e8a466 100644 --- a/README.md +++ b/README.md @@ -9,30 +9,36 @@ pip install getmnemo ## Quickstart ```python -from mnemo import Mnemo +from getmnemo import Mnemo -memory = Mnemo(api_key="lk_live_...", workspace_id="ws_...") +memory = Mnemo(api_key="prfly_live_...", default_container_tag="user:jane") # Store an atomic fact memory.add("User prefers Japanese short-grain rice for onigiri.") -# Retrieve relevant facts -hits = memory.search("what kind of rice does the user like?") -for hit in hits.hits: +# Retrieve relevant facts — returns a list of ranked hits +results = memory.search("what kind of rice does the user like?") +for hit in results: print(f"{hit.score:.2f} {hit.content}") ``` +Every call is scoped to a **container** — a `":"` string such as +`"user:jane"` or `"project:alpha"`. Set it once on the client +(`default_container_tag=`) or per call (`container_tag=`). Searching a +container that has never been written to returns an empty list. + Async variant: ```python import asyncio -from mnemo import AsyncMnemo +from getmnemo import AsyncMnemo async def main() -> None: - async with AsyncMnemo(api_key="...", workspace_id="...") as m: + async with AsyncMnemo(api_key="...", default_container_tag="user:jane") as m: await m.add("Trip to Costa Rica was 5 days, brought 7 shirts.") - res = await m.search("how many shirts did I pack?") - print(res.hits[0].content) + results = await m.search("how many shirts did I pack?") + if results: + print(results[0].content) asyncio.run(main()) ``` @@ -44,22 +50,42 @@ The client reads from env vars when arguments are not passed explicitly: | Env var | Default | Notes | |---|---|---| | `GETMNEMO_API_KEY` | (required) | from | -| `GETMNEMO_WORKSPACE_ID` | (required) | from the dashboard URL | -| `GETMNEMO_ACTOR_ID` | none | optional — scopes calls to a single user | +| `GETMNEMO_CONTAINER_TAG` | none | default container, e.g. `user:jane` | +| `GETMNEMO_WORKSPACE_ID` | none | deprecated — identity comes from the API key | | `GETMNEMO_API_URL` | `https://api.mnemohq.com` | override for self-hosted | ## API surface | Method | Purpose | |---|---| -| `search(query, *, limit=8, actor_id=None)` | Hybrid 7-strategy retrieval. Returns `SearchResponse`. | -| `add(content, *, metadata=None, actor_id=None)` | Store an atomic fact. Returns `Memory`. | -| `update(memory_id, *, content=None, metadata=None)` | Patch existing memory. | -| `delete(memory_id)` | Remove a memory. | -| `list(*, limit=20, cursor=None, actor_id=None)` | Cursor-paginated list. | +| `search(q, *, container_tag=None, limit=None)` | Hybrid retrieval. Returns `list[SearchHit]`. | +| `add(content, *, container_tag=None, metadata=None)` | Store an atomic fact. Returns `AddResponse`. | +| `add_many(items, *, container_tag=None)` | Store 1–100 memories in one request. Returns `AddResponse`. | +| `update(memory_id, *, content=None, metadata=None, container_tag=None)` | Patch existing memory. | +| `delete(memory_id, *, container_tag=None)` | Remove a memory. | +| `list(*, limit=20, cursor=None, container_tag=None)` | Cursor-paginated list. | All methods exist on both `Mnemo` (sync) and `AsyncMnemo` (async). +`add_many` items are content strings or mappings: + +```python +memory.add_many([ + "Standup moved to 9:30.", + {"content": "Invoices are net-30.", "memory_type": "fact", "idempotency_key": "inv-policy-1"}, +]) +``` + +### Upgrading from 0.1.x + +0.2.0 tracks the current Mnemo API; 0.1.x requests are rejected by it. + +- `workspace_id` is no longer required (the API key carries identity); pass a + container instead: `Mnemo(api_key=..., default_container_tag="user:jane")`. +- `search("q")` now returns a `list[SearchHit]` (was `SearchResponse.hits`). +- `add(...)` now returns an `AddResponse` (was a single `Memory`). +- `actor_id` was removed — use per-user container tags. + ## Development ```bash diff --git a/pyproject.toml b/pyproject.toml index 81d4d96..c39a918 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "hatchling.build" [project] name = "getmnemo" -version = "0.1.2" +version = "0.2.0" description = "Python SDK for Mnemo Memory — long-term memory infrastructure for AI agents." readme = "README.md" license = { text = "MIT" } @@ -45,11 +45,11 @@ dev = [ [project.urls] Homepage = "https://mnemohq.com" Documentation = "https://mnemohq.com/docs" -Repository = "https://github.com/getmnemo/getmnemo-python" -Issues = "https://github.com/getmnemo/getmnemo-python/issues" +Repository = "https://github.com/ledgermem/getmnemo-python" +Issues = "https://github.com/ledgermem/getmnemo-python/issues" [tool.hatch.build.targets.wheel] -packages = ["src/mnemo"] +packages = ["src/getmnemo", "src/mnemo"] [tool.ruff] line-length = 100 diff --git a/src/getmnemo/__init__.py b/src/getmnemo/__init__.py new file mode 100644 index 0000000..e6c6b41 --- /dev/null +++ b/src/getmnemo/__init__.py @@ -0,0 +1,54 @@ +"""Mnemo Memory — Python SDK. + +Quickstart: + + from getmnemo import Mnemo + + memory = Mnemo(api_key="prfly_live_...", default_container_tag="user:jane") + memory.add("User prefers Japanese short-grain rice for onigiri.") + results = memory.search("what kind of rice does the user like?") + for hit in results: + print(hit.score, hit.content) + +Async variant: + + import asyncio + from getmnemo import AsyncMnemo + + async def main() -> None: + async with AsyncMnemo(api_key="...", default_container_tag="user:jane") as m: + await m.add("Trip to Costa Rica was 5 days, brought 7 shirts.") + results = await m.search("how many shirts did I pack for Costa Rica?") + if results: + print(results[0].content) + + asyncio.run(main()) +""" + +from .client import AsyncMnemo, Mnemo +from .errors import MnemoError, MnemoHTTPError +from .models import ( + AddResponse, + Memory, + MemoryWriteReceipt, + MemoryWriteReceiptItem, + MemoryWriteStats, + PaginatedMemories, + SearchHit, +) + +__all__ = [ + "AddResponse", + "AsyncMnemo", + "Memory", + "MemoryWriteReceipt", + "MemoryWriteReceiptItem", + "MemoryWriteStats", + "Mnemo", + "MnemoError", + "MnemoHTTPError", + "PaginatedMemories", + "SearchHit", +] + +__version__ = "0.2.0" diff --git a/src/getmnemo/client.py b/src/getmnemo/client.py new file mode 100644 index 0000000..79ff82a --- /dev/null +++ b/src/getmnemo/client.py @@ -0,0 +1,562 @@ +"""Sync + async Mnemo clients. + +Both clients share the same surface; the async one is suitable for use inside +LangGraph nodes, FastAPI routes, or any other event-loop-friendly codebase. + +Wire contract (mirrors the TypeScript ``getmnemo`` 0.6 SDK): + +- ``POST /v1/memories`` body ``{containerTag, items: [{content, metadata?}]}`` +- ``POST /v1/search`` body ``{q, containerTag, limit?}`` -> ``{results: [...]}`` + +Every call needs a container tag of the form ``":"`` (for example +``"user:jane"``) — per call, via ``default_container_tag``, or via the +``GETMNEMO_CONTAINER_TAG`` environment variable. +""" + +from __future__ import annotations + +import asyncio +import os +import random +import time +from collections.abc import Awaitable, Mapping, Sequence +from typing import Any, Callable, TypeVar, Union +from urllib.parse import quote + +import httpx +from typing_extensions import Self, TypeAlias + +from .errors import MnemoHTTPError +from .models import AddResponse, Memory, PaginatedMemories, SearchHit + +_DEFAULT_BASE_URL = "https://api.mnemohq.com" +_DEFAULT_TIMEOUT = 30.0 +_USER_AGENT = "getmnemo-python/0.2.0" +_DEFAULT_MAX_RETRIES = 3 +_RETRY_BASE_DELAY = 0.2 +_RETRY_MAX_DELAY = 5.0 +_MAX_BATCH_SIZE = 100 + +# snake_case input keys accepted in add_many items -> camelCase wire keys. +_ITEM_KEY_MAP = { + "content": "content", + "metadata": "metadata", + "idempotency_key": "idempotencyKey", + "memory_type": "memoryType", +} + +T = TypeVar("T") +MemoryItemInput: TypeAlias = Union[str, Mapping[str, Any]] + + +def _retry_delay(attempt: int) -> float: + """Exponential backoff with full jitter.""" + capped = min(_RETRY_BASE_DELAY * (2**attempt), _RETRY_MAX_DELAY) + return random.uniform(0, capped) + + +def _is_retryable_status(status: int) -> bool: + # 501 Not Implemented is a permanent failure — retrying wastes round-trips. + if status == 501: + return False + return status == 429 or 500 <= status < 600 + + +def _retry_after_seconds(resp: httpx.Response) -> float | None: + """Parse a Retry-After header (seconds or HTTP-date). Returns ``None`` if absent or invalid.""" + raw = resp.headers.get("retry-after") + if not raw: + return None + raw = raw.strip() + try: + # Delta-seconds form. + secs = float(raw) + return max(0.0, min(secs, _RETRY_MAX_DELAY)) + except ValueError: + pass + try: + from datetime import datetime, timezone + from email.utils import parsedate_to_datetime + + when = parsedate_to_datetime(raw) + if when is None: + return None + if when.tzinfo is None: + when = when.replace(tzinfo=timezone.utc) + delta = (when - datetime.now(timezone.utc)).total_seconds() + return max(0.0, min(delta, _RETRY_MAX_DELAY)) + except (TypeError, ValueError): + return None + + +def _delay_for(resp: httpx.Response, attempt: int) -> float: + server_hint = _retry_after_seconds(resp) + if server_hint is not None: + return server_hint + return _retry_delay(attempt) + + +def _send_with_retries( + fn: Callable[[], httpx.Response], + max_retries: int, +) -> httpx.Response: + attempt = 0 + while True: + try: + resp = fn() + except (httpx.TransportError, httpx.TimeoutException): + if attempt >= max_retries: + raise + time.sleep(_retry_delay(attempt)) + attempt += 1 + continue + if _is_retryable_status(resp.status_code) and attempt < max_retries: + time.sleep(_delay_for(resp, attempt)) + attempt += 1 + continue + return resp + + +async def _send_with_retries_async( + fn: Callable[[], Awaitable[httpx.Response]], + max_retries: int, +) -> httpx.Response: + attempt = 0 + while True: + try: + resp = await fn() + except (httpx.TransportError, httpx.TimeoutException): + if attempt >= max_retries: + raise + await asyncio.sleep(_retry_delay(attempt)) + attempt += 1 + continue + if _is_retryable_status(resp.status_code) and attempt < max_retries: + await asyncio.sleep(_delay_for(resp, attempt)) + attempt += 1 + continue + return resp + + +def _resolve_api_key(value: str | None) -> str: + resolved = value or os.environ.get("GETMNEMO_API_KEY") + if not resolved: + raise ValueError( + "api_key is required. Pass it explicitly or set $GETMNEMO_API_KEY.", + ) + return resolved + + +def _headers(api_key: str, workspace_id: str | None) -> dict[str, str]: + h = { + "authorization": f"Bearer {api_key}", + "user-agent": _USER_AGENT, + "content-type": "application/json", + } + if workspace_id: + # Deprecated: workspace identity now comes from the API key. Sent only + # when explicitly configured, for backwards compatibility. + h["x-workspace-id"] = workspace_id + return h + + +def _raise_for_status(resp: httpx.Response) -> None: + if resp.is_success: + return + body: Any + try: + body = resp.json() + message = body.get("message") if isinstance(body, dict) else str(body) + except ValueError: + body = resp.text + message = body or resp.reason_phrase + raise MnemoHTTPError(message or "Unknown error", resp.status_code, body) + + +def _build_items(items: Sequence[MemoryItemInput]) -> list[dict[str, Any]]: + """Normalize add_many items to the wire shape, rejecting unknown keys. + + The API's global validation rejects unknown body keys with a 400, so we + fail fast client-side with a clearer message. + """ + if not 1 <= len(items) <= _MAX_BATCH_SIZE: + raise ValueError(f"add_many() takes between 1 and {_MAX_BATCH_SIZE} items") + built: list[dict[str, Any]] = [] + for index, item in enumerate(items): + if isinstance(item, str): + built.append({"content": item}) + continue + unknown = set(item) - set(_ITEM_KEY_MAP) + if unknown: + raise ValueError( + f"add_many() item {index} has unknown key(s) {sorted(unknown)}; " + f"allowed keys: {sorted(_ITEM_KEY_MAP)}" + ) + if not item.get("content"): + raise ValueError(f"add_many() item {index} requires a non-empty 'content'") + built.append( + {_ITEM_KEY_MAP[key]: value for key, value in item.items() if value is not None} + ) + return built + + +def _search_payload(q: str, container_tag: str, limit: int | None) -> dict[str, Any]: + payload: dict[str, Any] = {"q": q, "containerTag": container_tag} + if limit is not None: + payload["limit"] = limit + return payload + + +def _parse_search_results(data: Any) -> list[SearchHit]: + results = data.get("results") if isinstance(data, dict) else None + if not isinstance(results, list): + return [] + return [SearchHit.model_validate(hit) for hit in results] + + +class _BaseClient: + """Configuration shared by the sync and async clients.""" + + def __init__( + self, + api_key: str | None, + *, + default_container_tag: str | None, + workspace_id: str | None, + base_url: str | None, + max_retries: int, + ) -> None: + self._api_key = _resolve_api_key(api_key) + self._default_container_tag = default_container_tag or os.environ.get( + "GETMNEMO_CONTAINER_TAG" + ) + self._workspace_id = workspace_id or os.environ.get("GETMNEMO_WORKSPACE_ID") + self._base_url = ( + base_url or os.environ.get("GETMNEMO_API_URL") or _DEFAULT_BASE_URL + ).rstrip("/") + self._max_retries = max(0, int(max_retries)) + + def _require_container(self, method: str, container_tag: str | None) -> str: + tag = container_tag or self._default_container_tag + if not tag: + raise ValueError( + f"{method}: a container is required — pass container_tag " + '(a ":" string such as "user:jane") per call, set ' + "default_container_tag on the client, or set $GETMNEMO_CONTAINER_TAG." + ) + return tag + + def _container_params(self, container_tag: str | None) -> dict[str, str]: + """Optional container for the by-id and list routes. + + These routes stay callable without a container so the API remains the + single source of truth on whether one is mandatory. + """ + tag = container_tag or self._default_container_tag + return {"containerTag": tag} if tag else {} + + +class Mnemo(_BaseClient): + """Synchronous Mnemo Memory client. + + All public methods make a single HTTP round-trip and return typed Pydantic + models. The underlying httpx.Client is created on first use and reused. + """ + + def __init__( + self, + api_key: str | None = None, + *, + default_container_tag: str | None = None, + workspace_id: str | None = None, + base_url: str | None = None, + timeout: float = _DEFAULT_TIMEOUT, + client: httpx.Client | None = None, + max_retries: int = _DEFAULT_MAX_RETRIES, + ) -> None: + super().__init__( + api_key, + default_container_tag=default_container_tag, + workspace_id=workspace_id, + base_url=base_url, + max_retries=max_retries, + ) + self._owns_client = client is None + self._client = client or httpx.Client( + base_url=self._base_url, + timeout=timeout, + headers=_headers(self._api_key, self._workspace_id), + ) + + def __enter__(self) -> Self: + return self + + def __exit__(self, *_exc: object) -> None: + self.close() + + def close(self) -> None: + if self._owns_client: + self._client.close() + + # ----- public API ----- + + def search( + self, + q: str, + *, + container_tag: str | None = None, + limit: int | None = None, + ) -> list[SearchHit]: + """Hybrid retrieval. Returns the ranked results list. + + A container that has never been written to currently yields a 404 + ("Container not found"); that is treated as empty results. An upcoming + API change returns an empty list instead — both shapes are handled. + """ + tag = self._require_container("search()", container_tag) + resp = _send_with_retries( + lambda: self._client.post("/v1/search", json=_search_payload(q, tag, limit)), + self._max_retries, + ) + if resp.status_code == 404: + return [] + _raise_for_status(resp) + return _parse_search_results(resp.json()) + + def add( + self, + content: str, + *, + container_tag: str | None = None, + metadata: dict[str, Any] | None = None, + ) -> AddResponse: + """Store a single atomic fact.""" + item: dict[str, Any] = {"content": content} + if metadata is not None: + item["metadata"] = metadata + return self.add_many([item], container_tag=container_tag) + + def add_many( + self, + items: Sequence[MemoryItemInput], + *, + container_tag: str | None = None, + ) -> AddResponse: + """Store between 1 and 100 memories in one request. + + Each item is a content string or a mapping with ``content`` plus + optional ``metadata``, ``idempotency_key``, and ``memory_type``. + """ + tag = self._require_container("add_many()", container_tag) + payload = {"containerTag": tag, "items": _build_items(items)} + resp = _send_with_retries( + lambda: self._client.post("/v1/memories", json=payload), + self._max_retries, + ) + _raise_for_status(resp) + return AddResponse.model_validate(resp.json()) + + def update( + self, + memory_id: str, + *, + content: str | None = None, + metadata: dict[str, Any] | None = None, + container_tag: str | None = None, + ) -> Memory: + payload: dict[str, Any] = {} + if content is not None: + payload["content"] = content + if metadata is not None: + payload["metadata"] = metadata + if not payload: + raise ValueError("update() requires at least one of content/metadata") + resp = _send_with_retries( + lambda: self._client.patch( + f"/v1/memories/{quote(memory_id, safe='')}", + json=payload, + params=self._container_params(container_tag), + ), + self._max_retries, + ) + _raise_for_status(resp) + return Memory.model_validate(resp.json()) + + def delete(self, memory_id: str, *, container_tag: str | None = None) -> None: + resp = _send_with_retries( + lambda: self._client.delete( + f"/v1/memories/{quote(memory_id, safe='')}", + params=self._container_params(container_tag), + ), + self._max_retries, + ) + _raise_for_status(resp) + + def list( + self, + *, + limit: int = 20, + cursor: str | None = None, + container_tag: str | None = None, + ) -> PaginatedMemories: + params: dict[str, Any] = {"limit": limit, **self._container_params(container_tag)} + if cursor is not None: + params["cursor"] = cursor + resp = _send_with_retries( + lambda: self._client.get("/v1/memories", params=params), + self._max_retries, + ) + _raise_for_status(resp) + return PaginatedMemories.model_validate(resp.json()) + + +class AsyncMnemo(_BaseClient): + """Asynchronous variant of :class:`Mnemo`. + + Use as an async context manager so the underlying connection pool is + closed deterministically. + """ + + def __init__( + self, + api_key: str | None = None, + *, + default_container_tag: str | None = None, + workspace_id: str | None = None, + base_url: str | None = None, + timeout: float = _DEFAULT_TIMEOUT, + client: httpx.AsyncClient | None = None, + max_retries: int = _DEFAULT_MAX_RETRIES, + ) -> None: + super().__init__( + api_key, + default_container_tag=default_container_tag, + workspace_id=workspace_id, + base_url=base_url, + max_retries=max_retries, + ) + self._owns_client = client is None + self._client = client or httpx.AsyncClient( + base_url=self._base_url, + timeout=timeout, + headers=_headers(self._api_key, self._workspace_id), + ) + + async def __aenter__(self) -> Self: + return self + + async def __aexit__(self, *_exc: object) -> None: + await self.aclose() + + async def aclose(self) -> None: + if self._owns_client: + await self._client.aclose() + + async def search( + self, + q: str, + *, + container_tag: str | None = None, + limit: int | None = None, + ) -> list[SearchHit]: + """Hybrid retrieval. Returns the ranked results list. + + A container that has never been written to currently yields a 404 + ("Container not found"); that is treated as empty results. An upcoming + API change returns an empty list instead — both shapes are handled. + """ + tag = self._require_container("search()", container_tag) + resp = await _send_with_retries_async( + lambda: self._client.post("/v1/search", json=_search_payload(q, tag, limit)), + self._max_retries, + ) + if resp.status_code == 404: + return [] + _raise_for_status(resp) + return _parse_search_results(resp.json()) + + async def add( + self, + content: str, + *, + container_tag: str | None = None, + metadata: dict[str, Any] | None = None, + ) -> AddResponse: + """Store a single atomic fact.""" + item: dict[str, Any] = {"content": content} + if metadata is not None: + item["metadata"] = metadata + return await self.add_many([item], container_tag=container_tag) + + async def add_many( + self, + items: Sequence[MemoryItemInput], + *, + container_tag: str | None = None, + ) -> AddResponse: + """Store between 1 and 100 memories in one request. + + Each item is a content string or a mapping with ``content`` plus + optional ``metadata``, ``idempotency_key``, and ``memory_type``. + """ + tag = self._require_container("add_many()", container_tag) + payload = {"containerTag": tag, "items": _build_items(items)} + resp = await _send_with_retries_async( + lambda: self._client.post("/v1/memories", json=payload), + self._max_retries, + ) + _raise_for_status(resp) + return AddResponse.model_validate(resp.json()) + + async def update( + self, + memory_id: str, + *, + content: str | None = None, + metadata: dict[str, Any] | None = None, + container_tag: str | None = None, + ) -> Memory: + payload: dict[str, Any] = {} + if content is not None: + payload["content"] = content + if metadata is not None: + payload["metadata"] = metadata + if not payload: + raise ValueError("update() requires at least one of content/metadata") + resp = await _send_with_retries_async( + lambda: self._client.patch( + f"/v1/memories/{quote(memory_id, safe='')}", + json=payload, + params=self._container_params(container_tag), + ), + self._max_retries, + ) + _raise_for_status(resp) + return Memory.model_validate(resp.json()) + + async def delete(self, memory_id: str, *, container_tag: str | None = None) -> None: + resp = await _send_with_retries_async( + lambda: self._client.delete( + f"/v1/memories/{quote(memory_id, safe='')}", + params=self._container_params(container_tag), + ), + self._max_retries, + ) + _raise_for_status(resp) + + async def list( + self, + *, + limit: int = 20, + cursor: str | None = None, + container_tag: str | None = None, + ) -> PaginatedMemories: + params: dict[str, Any] = {"limit": limit, **self._container_params(container_tag)} + if cursor is not None: + params["cursor"] = cursor + resp = await _send_with_retries_async( + lambda: self._client.get("/v1/memories", params=params), + self._max_retries, + ) + _raise_for_status(resp) + return PaginatedMemories.model_validate(resp.json()) diff --git a/src/mnemo/errors.py b/src/getmnemo/errors.py similarity index 100% rename from src/mnemo/errors.py rename to src/getmnemo/errors.py diff --git a/src/getmnemo/models.py b/src/getmnemo/models.py new file mode 100644 index 0000000..3f1c773 --- /dev/null +++ b/src/getmnemo/models.py @@ -0,0 +1,92 @@ +"""Pydantic models mirroring the Mnemo REST API. + +Field names are snake_case in Python and map onto the camelCase wire format +via aliases. Models ignore unknown response fields so additive API changes +never break parsing. +""" + +from __future__ import annotations + +from datetime import datetime +from typing import Any + +from pydantic import BaseModel, ConfigDict, Field + + +class _Base(BaseModel): + model_config = ConfigDict(extra="ignore", populate_by_name=True) + + +class Memory(_Base): + """A stored memory record (``MemoryRecordDto``).""" + + id: str + content: str + scope_key: str | None = Field(default=None, alias="scopeKey") + memory_type: str | None = Field(default=None, alias="memoryType") + metadata: dict[str, Any] | None = None + source: dict[str, Any] | None = None + idempotency_key: str | None = Field(default=None, alias="idempotencyKey") + source_document_id: str | None = Field(default=None, alias="sourceDocumentId") + due_at: datetime | None = Field(default=None, alias="dueAt") + created_at: datetime | None = Field(default=None, alias="createdAt") + updated_at: datetime | None = Field(default=None, alias="updatedAt") + deleted_at: datetime | None = Field(default=None, alias="deletedAt") + + +class MemoryWriteStats(_Base): + total: int + created: int + deduplicated: int + + +class MemoryWriteReceiptItem(_Base): + input_index: int = Field(alias="inputIndex") + memory_id: str = Field(alias="memoryId") + status: str + search_status: str | None = Field(default=None, alias="searchStatus") + + +class MemoryWriteReceipt(_Base): + write_id: str = Field(alias="writeId") + status: str + searchable_at: datetime | None = Field(default=None, alias="searchableAt") + items: list[MemoryWriteReceiptItem] = Field(default_factory=list) + + +class AddResponse(_Base): + """Response of ``POST /v1/memories`` (``CreateMemoriesResponseDto``).""" + + scope_key: str = Field(alias="scopeKey") + items: list[Memory] + stats: MemoryWriteStats | None = None + receipt: MemoryWriteReceipt | None = None + + +class SearchHit(_Base): + """One ranked result from ``POST /v1/search`` (``SearchResultDto``).""" + + result_type: str = Field(alias="resultType") + score: float + scope_key: str = Field(alias="scopeKey") + content: str + metadata: dict[str, Any] | None = None + created_at: datetime | None = Field(default=None, alias="createdAt") + # Set when result_type is "memory". + memory_id: str | None = Field(default=None, alias="memoryId") + # Set when result_type is "document", and on memories derived from one. + document_id: str | None = Field(default=None, alias="documentId") + # Set when result_type is "fact". + fact_id: str | None = Field(default=None, alias="factId") + memory_type: str | None = Field(default=None, alias="memoryType") + content_type: str | None = Field(default=None, alias="contentType") + polarity: str | None = None + updated_at: datetime | None = Field(default=None, alias="updatedAt") + # Provenance / per-strategy scores; present only with includeSources=true. + sources: dict[str, Any] | None = None + + +class PaginatedMemories(_Base): + items: list[Memory] + next_cursor: str | None = Field(default=None, alias="nextCursor") + total: int | None = None diff --git a/src/getmnemo/py.typed b/src/getmnemo/py.typed new file mode 100644 index 0000000..e69de29 diff --git a/src/mnemo/__init__.py b/src/mnemo/__init__.py index 6a323d9..e1730d9 100644 --- a/src/mnemo/__init__.py +++ b/src/mnemo/__init__.py @@ -1,41 +1,35 @@ -"""Mnemo Memory — Python SDK. +"""Backwards-compatible alias for the ``getmnemo`` package. -Quickstart: - - from getmnemo import Mnemo - - memory = Mnemo(api_key="lk_live_...", workspace_id="ws_...") - memory.add("User prefers Japanese short-grain rice for onigiri.") - response = memory.search("what kind of rice does the user like?") - for hit in response.hits: - print(hit.score, hit.content) - -Async variant: - - import asyncio - from getmnemo import AsyncMnemo - - async def main() -> None: - async with AsyncMnemo(api_key="...", workspace_id="...") as m: - await m.add("Trip to Costa Rica was 5 days, brought 7 shirts.") - response = await m.search("how many shirts did I pack for Costa Rica?") - print(response.hits[0].content) - - asyncio.run(main()) +0.1.x shipped the module as ``mnemo``; the canonical import name now matches +the install name: ``from getmnemo import Mnemo``. """ -from .client import AsyncMnemo, Mnemo -from .errors import MnemoError, MnemoHTTPError -from .models import Memory, SearchHit, SearchResponse +from getmnemo import ( + AddResponse, + AsyncMnemo, + Memory, + MemoryWriteReceipt, + MemoryWriteReceiptItem, + MemoryWriteStats, + Mnemo, + MnemoError, + MnemoHTTPError, + PaginatedMemories, + SearchHit, + __version__, +) __all__ = [ + "AddResponse", "AsyncMnemo", + "Memory", + "MemoryWriteReceipt", + "MemoryWriteReceiptItem", + "MemoryWriteStats", "Mnemo", "MnemoError", "MnemoHTTPError", - "Memory", + "PaginatedMemories", "SearchHit", - "SearchResponse", + "__version__", ] - -__version__ = "0.1.0" diff --git a/src/mnemo/client.py b/src/mnemo/client.py deleted file mode 100644 index b1b9a48..0000000 --- a/src/mnemo/client.py +++ /dev/null @@ -1,402 +0,0 @@ -"""Sync + async Mnemo clients. - -Both clients share the same surface; the async one is suitable for use inside -LangGraph nodes, FastAPI routes, or any other event-loop-friendly codebase. -""" - -from __future__ import annotations - -import asyncio -import os -import random -import time -from typing import Any, Awaitable, Callable, TypeVar -from urllib.parse import quote - -import httpx -from typing_extensions import Self - -from .errors import MnemoHTTPError -from .models import Memory, PaginatedMemories, SearchResponse - -_DEFAULT_BASE_URL = "https://api.mnemohq.com" -_DEFAULT_TIMEOUT = 30.0 -_USER_AGENT = "getmnemo-python/0.1.0" -_DEFAULT_MAX_RETRIES = 3 -_RETRY_BASE_DELAY = 0.2 -_RETRY_MAX_DELAY = 5.0 - -T = TypeVar("T") - - -def _retry_delay(attempt: int) -> float: - """Exponential backoff with full jitter.""" - capped = min(_RETRY_BASE_DELAY * (2**attempt), _RETRY_MAX_DELAY) - return random.uniform(0, capped) - - -def _is_retryable_status(status: int) -> bool: - # 501 Not Implemented is a permanent failure — retrying wastes round-trips. - if status == 501: - return False - return status == 429 or 500 <= status < 600 - - -def _retry_after_seconds(resp: httpx.Response) -> float | None: - """Parse a Retry-After header (seconds or HTTP-date). Returns ``None`` if absent or invalid.""" - raw = resp.headers.get("retry-after") - if not raw: - return None - raw = raw.strip() - try: - # Delta-seconds form. - secs = float(raw) - return max(0.0, min(secs, _RETRY_MAX_DELAY)) - except ValueError: - pass - try: - from email.utils import parsedate_to_datetime - from datetime import datetime, timezone - - when = parsedate_to_datetime(raw) - if when is None: - return None - if when.tzinfo is None: - when = when.replace(tzinfo=timezone.utc) - delta = (when - datetime.now(timezone.utc)).total_seconds() - return max(0.0, min(delta, _RETRY_MAX_DELAY)) - except (TypeError, ValueError): - return None - - -def _delay_for(resp: httpx.Response, attempt: int) -> float: - server_hint = _retry_after_seconds(resp) - if server_hint is not None: - return server_hint - return _retry_delay(attempt) - - -def _send_with_retries( - fn: Callable[[], httpx.Response], - max_retries: int, -) -> httpx.Response: - attempt = 0 - while True: - try: - resp = fn() - except (httpx.TransportError, httpx.TimeoutException): - if attempt >= max_retries: - raise - time.sleep(_retry_delay(attempt)) - attempt += 1 - continue - if _is_retryable_status(resp.status_code) and attempt < max_retries: - time.sleep(_delay_for(resp, attempt)) - attempt += 1 - continue - return resp - - -async def _send_with_retries_async( - fn: Callable[[], Awaitable[httpx.Response]], - max_retries: int, -) -> httpx.Response: - attempt = 0 - while True: - try: - resp = await fn() - except (httpx.TransportError, httpx.TimeoutException): - if attempt >= max_retries: - raise - await asyncio.sleep(_retry_delay(attempt)) - attempt += 1 - continue - if _is_retryable_status(resp.status_code) and attempt < max_retries: - await asyncio.sleep(_delay_for(resp, attempt)) - attempt += 1 - continue - return resp - - -def _resolve(value: str | None, env_var: str, label: str) -> str: - resolved = value or os.environ.get(env_var) - if not resolved: - raise ValueError( - f"{label} is required. Pass it explicitly or set ${env_var}.", - ) - return resolved - - -def _headers(api_key: str, workspace_id: str, actor_id: str | None) -> dict[str, str]: - h = { - "authorization": f"Bearer {api_key}", - "x-workspace-id": workspace_id, - "user-agent": _USER_AGENT, - "content-type": "application/json", - } - if actor_id is not None: - h["x-actor-id"] = actor_id - return h - - -def _raise_for_status(resp: httpx.Response) -> None: - if resp.is_success: - return - body: Any - try: - body = resp.json() - message = body.get("message") if isinstance(body, dict) else str(body) - except ValueError: - body = resp.text - message = body or resp.reason_phrase - raise MnemoHTTPError(message or "Unknown error", resp.status_code, body) - - -class Mnemo: - """Synchronous Mnemo Memory client. - - All public methods make a single HTTP round-trip and return typed Pydantic - models. The underlying httpx.Client is created on first use and reused. - """ - - def __init__( - self, - api_key: str | None = None, - workspace_id: str | None = None, - *, - actor_id: str | None = None, - base_url: str | None = None, - timeout: float = _DEFAULT_TIMEOUT, - client: httpx.Client | None = None, - max_retries: int = _DEFAULT_MAX_RETRIES, - ) -> None: - self._api_key = _resolve(api_key, "GETMNEMO_API_KEY", "api_key") - self._workspace_id = _resolve(workspace_id, "GETMNEMO_WORKSPACE_ID", "workspace_id") - self._actor_id = actor_id or os.environ.get("GETMNEMO_ACTOR_ID") - self._base_url = (base_url or os.environ.get("GETMNEMO_API_URL") or _DEFAULT_BASE_URL).rstrip("/") - self._owns_client = client is None - self._max_retries = max(0, int(max_retries)) - self._client = client or httpx.Client( - base_url=self._base_url, - timeout=timeout, - headers=_headers(self._api_key, self._workspace_id, self._actor_id), - ) - - def __enter__(self) -> Self: - return self - - def __exit__(self, *_exc: object) -> None: - self.close() - - def close(self) -> None: - if self._owns_client: - self._client.close() - - # ----- public API ----- - - def search( - self, - query: str, - *, - limit: int = 8, - actor_id: str | None = None, - ) -> SearchResponse: - payload: dict[str, Any] = {"query": query, "limit": limit} - if actor_id is not None: - payload["actorId"] = actor_id - resp = _send_with_retries( - lambda: self._client.post("/v1/search", json=payload), - self._max_retries, - ) - _raise_for_status(resp) - return SearchResponse.model_validate(resp.json()) - - def add( - self, - content: str, - *, - metadata: dict[str, Any] | None = None, - actor_id: str | None = None, - ) -> Memory: - payload: dict[str, Any] = {"content": content} - if metadata is not None: - payload["metadata"] = metadata - if actor_id is not None: - payload["actorId"] = actor_id - resp = _send_with_retries( - lambda: self._client.post("/v1/memories", json=payload), - self._max_retries, - ) - _raise_for_status(resp) - return Memory.model_validate(resp.json()) - - def update( - self, - memory_id: str, - *, - content: str | None = None, - metadata: dict[str, Any] | None = None, - ) -> Memory: - payload: dict[str, Any] = {} - if content is not None: - payload["content"] = content - if metadata is not None: - payload["metadata"] = metadata - if not payload: - raise ValueError("update() requires at least one of content/metadata") - resp = _send_with_retries( - lambda: self._client.patch(f"/v1/memories/{quote(memory_id, safe='')}", json=payload), - self._max_retries, - ) - _raise_for_status(resp) - return Memory.model_validate(resp.json()) - - def delete(self, memory_id: str) -> None: - resp = _send_with_retries( - lambda: self._client.delete(f"/v1/memories/{quote(memory_id, safe='')}"), - self._max_retries, - ) - _raise_for_status(resp) - - def list( - self, - *, - limit: int = 20, - cursor: str | None = None, - actor_id: str | None = None, - ) -> PaginatedMemories: - params: dict[str, Any] = {"limit": limit} - if cursor is not None: - params["cursor"] = cursor - if actor_id is not None: - params["actorId"] = actor_id - resp = _send_with_retries( - lambda: self._client.get("/v1/memories", params=params), - self._max_retries, - ) - _raise_for_status(resp) - return PaginatedMemories.model_validate(resp.json()) - - -class AsyncMnemo: - """Asynchronous variant of :class:`Mnemo`. - - Use as an async context manager so the underlying connection pool is - closed deterministically. - """ - - def __init__( - self, - api_key: str | None = None, - workspace_id: str | None = None, - *, - actor_id: str | None = None, - base_url: str | None = None, - timeout: float = _DEFAULT_TIMEOUT, - client: httpx.AsyncClient | None = None, - max_retries: int = _DEFAULT_MAX_RETRIES, - ) -> None: - self._api_key = _resolve(api_key, "GETMNEMO_API_KEY", "api_key") - self._workspace_id = _resolve(workspace_id, "GETMNEMO_WORKSPACE_ID", "workspace_id") - self._actor_id = actor_id or os.environ.get("GETMNEMO_ACTOR_ID") - self._base_url = (base_url or os.environ.get("GETMNEMO_API_URL") or _DEFAULT_BASE_URL).rstrip("/") - self._owns_client = client is None - self._max_retries = max(0, int(max_retries)) - self._client = client or httpx.AsyncClient( - base_url=self._base_url, - timeout=timeout, - headers=_headers(self._api_key, self._workspace_id, self._actor_id), - ) - - async def __aenter__(self) -> Self: - return self - - async def __aexit__(self, *_exc: object) -> None: - await self.aclose() - - async def aclose(self) -> None: - if self._owns_client: - await self._client.aclose() - - async def search( - self, - query: str, - *, - limit: int = 8, - actor_id: str | None = None, - ) -> SearchResponse: - payload: dict[str, Any] = {"query": query, "limit": limit} - if actor_id is not None: - payload["actorId"] = actor_id - resp = await _send_with_retries_async( - lambda: self._client.post("/v1/search", json=payload), - self._max_retries, - ) - _raise_for_status(resp) - return SearchResponse.model_validate(resp.json()) - - async def add( - self, - content: str, - *, - metadata: dict[str, Any] | None = None, - actor_id: str | None = None, - ) -> Memory: - payload: dict[str, Any] = {"content": content} - if metadata is not None: - payload["metadata"] = metadata - if actor_id is not None: - payload["actorId"] = actor_id - resp = await _send_with_retries_async( - lambda: self._client.post("/v1/memories", json=payload), - self._max_retries, - ) - _raise_for_status(resp) - return Memory.model_validate(resp.json()) - - async def update( - self, - memory_id: str, - *, - content: str | None = None, - metadata: dict[str, Any] | None = None, - ) -> Memory: - payload: dict[str, Any] = {} - if content is not None: - payload["content"] = content - if metadata is not None: - payload["metadata"] = metadata - if not payload: - raise ValueError("update() requires at least one of content/metadata") - resp = await _send_with_retries_async( - lambda: self._client.patch(f"/v1/memories/{quote(memory_id, safe='')}", json=payload), - self._max_retries, - ) - _raise_for_status(resp) - return Memory.model_validate(resp.json()) - - async def delete(self, memory_id: str) -> None: - resp = await _send_with_retries_async( - lambda: self._client.delete(f"/v1/memories/{quote(memory_id, safe='')}"), - self._max_retries, - ) - _raise_for_status(resp) - - async def list( - self, - *, - limit: int = 20, - cursor: str | None = None, - actor_id: str | None = None, - ) -> PaginatedMemories: - params: dict[str, Any] = {"limit": limit} - if cursor is not None: - params["cursor"] = cursor - if actor_id is not None: - params["actorId"] = actor_id - resp = await _send_with_retries_async( - lambda: self._client.get("/v1/memories", params=params), - self._max_retries, - ) - _raise_for_status(resp) - return PaginatedMemories.model_validate(resp.json()) diff --git a/src/mnemo/models.py b/src/mnemo/models.py deleted file mode 100644 index 1554e1e..0000000 --- a/src/mnemo/models.py +++ /dev/null @@ -1,46 +0,0 @@ -"""Pydantic models mirroring the Mnemo REST API.""" - -from __future__ import annotations - -from datetime import datetime -from typing import Any - -from pydantic import BaseModel, ConfigDict, Field - - -class _Base(BaseModel): - model_config = ConfigDict(extra="ignore", populate_by_name=True) - - -class Memory(_Base): - id: str - content: str - metadata: dict[str, Any] | None = None - workspace_id: str = Field(alias="workspaceId") - actor_id: str | None = Field(default=None, alias="actorId") - created_at: datetime = Field(alias="createdAt") - updated_at: datetime = Field(alias="updatedAt") - - -class SearchSource(_Base): - document_id: str | None = Field(default=None, alias="documentId") - chunk_id: str | None = Field(default=None, alias="chunkId") - - -class SearchHit(_Base): - memory_id: str = Field(alias="memoryId") - content: str - score: float - metadata: dict[str, Any] | None = None - source: SearchSource | None = None - - -class SearchResponse(_Base): - hits: list[SearchHit] - query: str - latency_ms: float = Field(alias="latencyMs", ge=0) - - -class PaginatedMemories(_Base): - items: list[Memory] - next_cursor: str | None = Field(default=None, alias="nextCursor") diff --git a/tests/test_client.py b/tests/test_client.py index b0e220c..045a4af 100644 --- a/tests/test_client.py +++ b/tests/test_client.py @@ -1,7 +1,8 @@ -"""Smoke tests for the Mnemo client. +"""Offline tests for the Mnemo client, using httpx's MockTransport. -These tests use httpx's MockTransport so they run offline. Real-API smoke -tests live in tests/integration/ and are gated by GETMNEMO_RUN_INTEGRATION=1. +They pin the wire contract: request bodies must contain ONLY the keys the API +accepts (the server's global validation rejects unknown keys with a 400), and +responses parse into the typed models. """ from __future__ import annotations @@ -15,119 +16,343 @@ from getmnemo import AsyncMnemo, Mnemo, MnemoHTTPError -ISO = datetime(2026, 4, 27, 12, 0, 0, tzinfo=timezone.utc).isoformat() +ISO = datetime(2026, 10, 14, 12, 0, 0, tzinfo=timezone.utc).isoformat() +# Keys the API accepts on each body. Anything else trips the global +# ValidationPipe with a 400 — the tests below assert we never send more. +ALLOWED_ADD_KEYS = {"scope", "containerTag", "containerType", "metadata", "source", "items"} +ALLOWED_ITEM_KEYS = { + "id", + "content", + "idempotencyKey", + "memoryType", + "polarity", + "dueAt", + "mutationPolicy", + "metadata", + "source", +} +ALLOWED_SEARCH_KEYS = { + "scope", + "containerTag", + "q", + "scopes", + "containerTags", + "searchMode", + "limit", + "filters", + "includeSources", + "polarity", + "intentOverride", + "agentic", + "mode", +} -def _memory_response(memory_id: str = "mem_123") -> dict[str, Any]: + +def _memory_record(memory_id: str = "11111111-1111-4111-8111-111111111111") -> dict[str, Any]: return { "id": memory_id, + "scopeKey": "user:jane", "content": "User prefers Japanese rice.", + "memoryType": "preference", "metadata": {"source": "test"}, - "workspaceId": "ws_test", - "actorId": None, "createdAt": ISO, "updatedAt": ISO, } -def _mock(handler: httpx.MockTransport) -> httpx.Client: - return httpx.Client( - base_url="http://test", - transport=handler, - headers={"authorization": "Bearer test"}, - ) +def _add_response() -> dict[str, Any]: + return { + "scopeKey": "user:jane", + "scope": {"type": "user", "id": "jane"}, + "items": [_memory_record()], + "stats": {"total": 1, "created": 1, "deduplicated": 0}, + "receipt": { + "writeId": "22222222-2222-4222-8222-222222222222", + "status": "searchable", + "searchableAt": ISO, + "items": [ + { + "inputIndex": 0, + "memoryId": "11111111-1111-4111-8111-111111111111", + "status": "created", + "searchStatus": "searchable", + } + ], + }, + } -def test_search_returns_typed_hits() -> None: +def _search_response() -> dict[str, Any]: + return { + "results": [ + { + "resultType": "memory", + "score": 0.91, + "scopeKey": "user:jane", + "content": "User prefers Japanese rice.", + "metadata": {"source": "test"}, + "createdAt": ISO, + "updatedAt": ISO, + "memoryId": "11111111-1111-4111-8111-111111111111", + "memoryType": "preference", + "polarity": "positive", + } + ], + "positivePreferences": [], + "hardConstraints": [], + "searchMode": "hybrid", + "queryIntent": "simple_recall", + "queryIntentConfidence": 1, + "abstained": False, + "reranked": False, + "rawBestVectorSim": 0.7, + "latency": { + "parallelMs": 1, + "strategyMs": 1, + "fusionMs": 0, + "rerankerMs": 0, + "totalMs": 2, + }, + } + + +def _sync(handler) -> Mnemo: # type: ignore[no-untyped-def] + client = httpx.Client(base_url="http://test", transport=httpx.MockTransport(handler)) + return Mnemo(api_key="test", default_container_tag="user:jane", client=client) + + +# ----- add ----- + + +def test_add_sends_container_and_items_only() -> None: def handler(req: httpx.Request) -> httpx.Response: - assert req.url.path == "/v1/search" + assert req.method == "POST" + assert req.url.path == "/v1/memories" body = json.loads(req.content) - assert body["query"] == "rice" - assert body["limit"] == 5 - return httpx.Response( - 200, - json={ - "hits": [ - { - "memoryId": "mem_1", - "content": "User prefers Japanese rice.", - "score": 0.91, - "metadata": {}, - "source": {"documentId": "doc_1", "chunkId": "ck_1"}, - } - ], - "query": "rice", - "latencyMs": 42, - }, - ) + assert set(body) <= ALLOWED_ADD_KEYS, f"unknown body keys: {set(body) - ALLOWED_ADD_KEYS}" + assert body["containerTag"] == "user:jane" + assert isinstance(body["items"], list) and len(body["items"]) == 1 + item = body["items"][0] + assert set(item) <= ALLOWED_ITEM_KEYS, f"unknown item keys: {set(item) - ALLOWED_ITEM_KEYS}" + assert item == { + "content": "User prefers Japanese rice.", + "metadata": {"source": "test"}, + } + return httpx.Response(201, json=_add_response()) + + res = _sync(handler).add("User prefers Japanese rice.", metadata={"source": "test"}) + assert res.scope_key == "user:jane" + assert res.items[0].id == "11111111-1111-4111-8111-111111111111" + assert res.stats is not None and res.stats.created == 1 + assert res.receipt is not None and res.receipt.items[0].status == "created" + +def test_add_per_call_container_overrides_default() -> None: + def handler(req: httpx.Request) -> httpx.Response: + body = json.loads(req.content) + assert body["containerTag"] == "project:alpha" + return httpx.Response(201, json=_add_response()) + + _sync(handler).add("note", container_tag="project:alpha") + + +def test_add_many_builds_items_and_rejects_unknown_keys() -> None: + def handler(req: httpx.Request) -> httpx.Response: + body = json.loads(req.content) + assert set(body) == {"containerTag", "items"} + assert body["items"] == [ + {"content": "a"}, + {"content": "b", "idempotencyKey": "k-1", "memoryType": "fact"}, + ] + return httpx.Response(201, json=_add_response()) + + client = _sync(handler) + client.add_many(["a", {"content": "b", "idempotency_key": "k-1", "memory_type": "fact"}]) + + with pytest.raises(ValueError, match="unknown key"): + client.add_many([{"content": "x", "actor_id": "nope"}]) + with pytest.raises(ValueError, match="between 1 and 100"): + client.add_many([]) + with pytest.raises(ValueError, match="between 1 and 100"): + client.add_many(["x"] * 101) + with pytest.raises(ValueError, match="content"): + client.add_many([{"metadata": {}}]) + + +def test_container_required_when_no_default() -> None: client = Mnemo( api_key="test", - workspace_id="ws_test", - client=_mock(httpx.MockTransport(handler)), + client=httpx.Client( + base_url="http://test", + transport=httpx.MockTransport(lambda _r: httpx.Response(201, json=_add_response())), + ), ) - res = client.search("rice", limit=5) - assert res.latency_ms == 42 - assert len(res.hits) == 1 - assert res.hits[0].score == 0.91 - assert res.hits[0].source is not None - assert res.hits[0].source.document_id == "doc_1" + with pytest.raises(ValueError, match=r":"): + client.add("orphan content") + with pytest.raises(ValueError, match=r":"): + client.search("anything") -def test_add_memory_returns_typed_memory() -> None: +def test_container_tag_env_fallback(monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setenv("GETMNEMO_CONTAINER_TAG", "user:envjane") + def handler(req: httpx.Request) -> httpx.Response: - assert req.method == "POST" - assert req.url.path == "/v1/memories" - return httpx.Response(200, json=_memory_response()) + body = json.loads(req.content) + assert body["containerTag"] == "user:envjane" + return httpx.Response(201, json=_add_response()) client = Mnemo( api_key="test", - workspace_id="ws_test", - client=_mock(httpx.MockTransport(handler)), + client=httpx.Client(base_url="http://test", transport=httpx.MockTransport(handler)), ) - mem = client.add("User prefers Japanese rice.", metadata={"source": "test"}) - assert mem.id == "mem_123" - assert mem.workspace_id == "ws_test" + client.add("env container") + + +# ----- search ----- + + +def test_search_sends_q_and_container_only() -> None: + def handler(req: httpx.Request) -> httpx.Response: + assert req.url.path == "/v1/search" + body = json.loads(req.content) + assert set(body) <= ALLOWED_SEARCH_KEYS, ( + f"unknown body keys: {set(body) - ALLOWED_SEARCH_KEYS}" + ) + assert body == {"q": "rice", "containerTag": "user:jane"} + return httpx.Response(201, json=_search_response()) + + results = _sync(handler).search("rice") + assert len(results) == 1 + hit = results[0] + assert hit.result_type == "memory" + assert hit.score == 0.91 + assert hit.memory_id == "11111111-1111-4111-8111-111111111111" + assert hit.content == "User prefers Japanese rice." + assert hit.polarity == "positive" + + +def test_search_sends_limit_when_given() -> None: + def handler(req: httpx.Request) -> httpx.Response: + body = json.loads(req.content) + assert body == {"q": "rice", "containerTag": "user:jane", "limit": 5} + return httpx.Response(201, json=_search_response()) + + _sync(handler).search("rice", limit=5) + + +def test_search_container_not_found_returns_empty() -> None: + def handler(_req: httpx.Request) -> httpx.Response: + return httpx.Response(404, json={"message": "Container not found", "statusCode": 404}) + + assert _sync(handler).search("anything") == [] + + +def test_search_tolerates_future_empty_results_shape() -> None: + # The in-flight API change returns 200 + empty results instead of 404. + def handler(_req: httpx.Request) -> httpx.Response: + return httpx.Response(200, json={"results": [], "abstained": True}) + + assert _sync(handler).search("anything") == [] + + +# ----- errors + workspace header ----- def test_http_error_includes_status_and_body() -> None: def handler(_req: httpx.Request) -> httpx.Response: return httpx.Response(401, json={"message": "invalid api key"}) - client = Mnemo( - api_key="test", - workspace_id="ws_test", - client=_mock(httpx.MockTransport(handler)), - ) with pytest.raises(MnemoHTTPError) as ei: - client.search("anything") + _sync(handler).search("anything") assert ei.value.status == 401 assert "invalid api key" in str(ei.value) -def test_update_requires_at_least_one_field() -> None: - client = Mnemo( - api_key="test", - workspace_id="ws_test", - client=_mock(httpx.MockTransport(lambda _r: httpx.Response(200, json=_memory_response()))), - ) +def test_add_404_still_raises() -> None: + # Only search() swallows 404s; writes must surface them. + def handler(_req: httpx.Request) -> httpx.Response: + return httpx.Response(404, json={"message": "not found"}) + + with pytest.raises(MnemoHTTPError) as ei: + _sync(handler).add("content") + assert ei.value.status == 404 + + +def test_workspace_header_only_sent_when_configured( + monkeypatch: pytest.MonkeyPatch, +) -> None: + monkeypatch.delenv("GETMNEMO_WORKSPACE_ID", raising=False) + without = Mnemo(api_key="test", default_container_tag="user:jane") + assert "x-workspace-id" not in without._client.headers + without.close() + + with_ws = Mnemo(api_key="test", default_container_tag="user:jane", workspace_id="ws_1") + assert with_ws._client.headers["x-workspace-id"] == "ws_1" + with_ws.close() + + +# ----- update / delete / list ----- + + +def test_update_threads_container_and_requires_a_field() -> None: + def handler(req: httpx.Request) -> httpx.Response: + assert req.method == "PATCH" + assert req.url.params["containerTag"] == "user:jane" + body = json.loads(req.content) + assert body == {"content": "updated"} + return httpx.Response(200, json=_memory_record()) + + client = _sync(handler) + mem = client.update("mem_1", content="updated") + assert mem.content == "User prefers Japanese rice." with pytest.raises(ValueError): client.update("mem_1") -@pytest.mark.asyncio -async def test_async_search_round_trips() -> None: +def test_list_threads_container() -> None: def handler(req: httpx.Request) -> httpx.Response: - assert req.url.path == "/v1/search" + assert req.url.params["containerTag"] == "user:jane" + assert req.url.params["limit"] == "20" return httpx.Response( - 200, - json={"hits": [], "query": "anything", "latencyMs": 1}, + 200, json={"items": [_memory_record()], "nextCursor": None, "total": 1} ) - transport = httpx.MockTransport(handler) - async_client = httpx.AsyncClient(base_url="http://test", transport=transport) - async with AsyncMnemo( - api_key="test", workspace_id="ws_test", client=async_client - ) as m: - res = await m.search("anything") - assert res.hits == [] + page = _sync(handler).list() + assert page.total == 1 + assert page.items[0].scope_key == "user:jane" + + +# ----- async ----- + + +def _async(handler) -> AsyncMnemo: # type: ignore[no-untyped-def] + client = httpx.AsyncClient(base_url="http://test", transport=httpx.MockTransport(handler)) + return AsyncMnemo(api_key="test", default_container_tag="user:jane", client=client) + + +@pytest.mark.asyncio +async def test_async_add_and_search_round_trip() -> None: + def handler(req: httpx.Request) -> httpx.Response: + body = json.loads(req.content) + if req.url.path == "/v1/memories": + assert set(body) == {"containerTag", "items"} + return httpx.Response(201, json=_add_response()) + assert req.url.path == "/v1/search" + assert body == {"q": "rice", "containerTag": "user:jane"} + return httpx.Response(201, json=_search_response()) + + async with _async(handler) as m: + added = await m.add("User prefers Japanese rice.") + assert added.items[0].content == "User prefers Japanese rice." + results = await m.search("rice") + assert results[0].score == 0.91 + + +@pytest.mark.asyncio +async def test_async_search_container_not_found_returns_empty() -> None: + def handler(_req: httpx.Request) -> httpx.Response: + return httpx.Response(404, json={"message": "Container not found"}) + + async with _async(handler) as m: + assert await m.search("anything") == [] From c73fe9536fe26255708560e9adb417f795834cc0 Mon Sep 17 00:00:00 2001 From: Shah Hussain <95882307+shhahhussain@users.noreply.github.com> Date: Thu, 15 Oct 2026 19:10:20 +0500 Subject: [PATCH 2/2] fix: narrow search 404 to container-not-found; auto idempotency key on retried writes - search (sync + async) returns [] only when a 404 body message says 'Container not found'; any other 404 raises MnemoHTTPError - add/add_many generate a per-item uuid4 idempotencyKey when the caller gave none and retries are enabled; the payload is built once, so every retry carries the same key (no duplicate writes on 5xx/429/transport retries) --- src/getmnemo/client.py | 50 +++++++++++++++++++++++---- tests/test_client.py | 76 ++++++++++++++++++++++++++++++++++++++++++ 2 files changed, 119 insertions(+), 7 deletions(-) diff --git a/src/getmnemo/client.py b/src/getmnemo/client.py index 79ff82a..f6e0ce7 100644 --- a/src/getmnemo/client.py +++ b/src/getmnemo/client.py @@ -19,6 +19,7 @@ import os import random import time +import uuid from collections.abc import Awaitable, Mapping, Sequence from typing import Any, Callable, TypeVar, Union from urllib.parse import quote @@ -173,7 +174,22 @@ def _raise_for_status(resp: httpx.Response) -> None: raise MnemoHTTPError(message or "Unknown error", resp.status_code, body) -def _build_items(items: Sequence[MemoryItemInput]) -> list[dict[str, Any]]: +def _is_container_not_found(resp: httpx.Response) -> bool: + """True only for the API's 404 "Container not found" (never-written container).""" + if resp.status_code != 404: + return False + try: + body = resp.json() + except ValueError: + return False + if not isinstance(body, dict): + return False + return "container not found" in str(body.get("message", "")).lower() + + +def _build_items( + items: Sequence[MemoryItemInput], *, auto_idempotency_key: bool = False +) -> list[dict[str, Any]]: """Normalize add_many items to the wire shape, rejecting unknown keys. The API's global validation rejects unknown body keys with a 400, so we @@ -184,7 +200,7 @@ def _build_items(items: Sequence[MemoryItemInput]) -> list[dict[str, Any]]: built: list[dict[str, Any]] = [] for index, item in enumerate(items): if isinstance(item, str): - built.append({"content": item}) + built.append(_with_key({"content": item}, auto_idempotency_key)) continue unknown = set(item) - set(_ITEM_KEY_MAP) if unknown: @@ -195,11 +211,25 @@ def _build_items(items: Sequence[MemoryItemInput]) -> list[dict[str, Any]]: if not item.get("content"): raise ValueError(f"add_many() item {index} requires a non-empty 'content'") built.append( - {_ITEM_KEY_MAP[key]: value for key, value in item.items() if value is not None} + _with_key( + {_ITEM_KEY_MAP[key]: value for key, value in item.items() if value is not None}, + auto_idempotency_key, + ) ) return built +def _with_key(item: dict[str, Any], auto: bool) -> dict[str, Any]: + """Add a generated idempotencyKey so a retried write cannot duplicate. + + Generated once per call (the payload is built before the retry loop), so + every retry of the same request carries the same key. + """ + if auto and "idempotencyKey" not in item: + return {**item, "idempotencyKey": str(uuid.uuid4())} + return item + + def _search_payload(q: str, container_tag: str, limit: int | None) -> dict[str, Any]: payload: dict[str, Any] = {"q": q, "containerTag": container_tag} if limit is not None: @@ -318,7 +348,7 @@ def search( lambda: self._client.post("/v1/search", json=_search_payload(q, tag, limit)), self._max_retries, ) - if resp.status_code == 404: + if _is_container_not_found(resp): return [] _raise_for_status(resp) return _parse_search_results(resp.json()) @@ -348,7 +378,10 @@ def add_many( optional ``metadata``, ``idempotency_key``, and ``memory_type``. """ tag = self._require_container("add_many()", container_tag) - payload = {"containerTag": tag, "items": _build_items(items)} + payload = { + "containerTag": tag, + "items": _build_items(items, auto_idempotency_key=self._max_retries > 0), + } resp = _send_with_retries( lambda: self._client.post("/v1/memories", json=payload), self._max_retries, @@ -470,7 +503,7 @@ async def search( lambda: self._client.post("/v1/search", json=_search_payload(q, tag, limit)), self._max_retries, ) - if resp.status_code == 404: + if _is_container_not_found(resp): return [] _raise_for_status(resp) return _parse_search_results(resp.json()) @@ -500,7 +533,10 @@ async def add_many( optional ``metadata``, ``idempotency_key``, and ``memory_type``. """ tag = self._require_container("add_many()", container_tag) - payload = {"containerTag": tag, "items": _build_items(items)} + payload = { + "containerTag": tag, + "items": _build_items(items, auto_idempotency_key=self._max_retries > 0), + } resp = await _send_with_retries_async( lambda: self._client.post("/v1/memories", json=payload), self._max_retries, diff --git a/tests/test_client.py b/tests/test_client.py index 045a4af..e293bae 100644 --- a/tests/test_client.py +++ b/tests/test_client.py @@ -135,6 +135,7 @@ def handler(req: httpx.Request) -> httpx.Response: assert isinstance(body["items"], list) and len(body["items"]) == 1 item = body["items"][0] assert set(item) <= ALLOWED_ITEM_KEYS, f"unknown item keys: {set(item) - ALLOWED_ITEM_KEYS}" + assert isinstance(item.pop("idempotencyKey"), str) # auto-generated assert item == { "content": "User prefers Japanese rice.", "metadata": {"source": "test"}, @@ -161,8 +162,11 @@ def test_add_many_builds_items_and_rejects_unknown_keys() -> None: def handler(req: httpx.Request) -> httpx.Response: body = json.loads(req.content) assert set(body) == {"containerTag", "items"} + auto_key = body["items"][0].pop("idempotencyKey") + assert isinstance(auto_key, str) and auto_key assert body["items"] == [ {"content": "a"}, + # Caller-supplied key is kept, not replaced. {"content": "b", "idempotencyKey": "k-1", "memoryType": "fact"}, ] return httpx.Response(201, json=_add_response()) @@ -248,6 +252,47 @@ def handler(_req: httpx.Request) -> httpx.Response: assert _sync(handler).search("anything") == [] +def test_search_other_404_raises() -> None: + def handler(_req: httpx.Request) -> httpx.Response: + return httpx.Response(404, json={"message": "Cannot POST /v1/serch", "statusCode": 404}) + + with pytest.raises(MnemoHTTPError) as exc: + _sync(handler).search("anything") + assert exc.value.status == 404 + + +def test_search_404_non_json_raises() -> None: + def handler(_req: httpx.Request) -> httpx.Response: + return httpx.Response(404, text="not found") + + with pytest.raises(MnemoHTTPError): + _sync(handler).search("anything") + + +def test_add_idempotency_key_stable_across_retries(monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setattr("getmnemo.client.time.sleep", lambda _s: None) + keys: list[str] = [] + + def handler(req: httpx.Request) -> httpx.Response: + keys.append(json.loads(req.content)["items"][0]["idempotencyKey"]) + if len(keys) == 1: + return httpx.Response(503, json={"message": "unavailable"}) + return httpx.Response(201, json=_add_response()) + + _sync(handler).add("note") + assert len(keys) == 2 + assert keys[0] == keys[1] and keys[0] + + +def test_add_no_auto_key_when_retries_disabled() -> None: + def handler(req: httpx.Request) -> httpx.Response: + assert "idempotencyKey" not in json.loads(req.content)["items"][0] + return httpx.Response(201, json=_add_response()) + + client = httpx.Client(base_url="http://test", transport=httpx.MockTransport(handler)) + Mnemo(api_key="t", default_container_tag="user:jane", client=client, max_retries=0).add("n") + + def test_search_tolerates_future_empty_results_shape() -> None: # The in-flight API change returns 200 + empty results instead of 404. def handler(_req: httpx.Request) -> httpx.Response: @@ -356,3 +401,34 @@ def handler(_req: httpx.Request) -> httpx.Response: async with _async(handler) as m: assert await m.search("anything") == [] + + +@pytest.mark.asyncio +async def test_async_search_other_404_raises() -> None: + def handler(_req: httpx.Request) -> httpx.Response: + return httpx.Response(404, json={"message": "Route not found"}) + + async with _async(handler) as m: + with pytest.raises(MnemoHTTPError): + await m.search("anything") + + +@pytest.mark.asyncio +async def test_async_add_idempotency_key_stable_across_retries( + monkeypatch: pytest.MonkeyPatch, +) -> None: + async def no_sleep(_s: float) -> None: + return None + + monkeypatch.setattr("getmnemo.client.asyncio.sleep", no_sleep) + keys: list[str] = [] + + def handler(req: httpx.Request) -> httpx.Response: + keys.append(json.loads(req.content)["items"][0]["idempotencyKey"]) + if len(keys) == 1: + return httpx.Response(429, json={"message": "slow down"}) + return httpx.Response(201, json=_add_response()) + + async with _async(handler) as m: + await m.add("note") + assert len(keys) == 2 and keys[0] == keys[1]