persistmemory

The official Python client for the PersistMemory API. Sync and async, typed, with a py.typed marker.

Quickstart

from persistmemory import PersistMemory

client = PersistMemory()  # reads PERSISTMEMORY_API_KEY

# A Space is required. There is no account default to fall back to — see below.
work = next(space for space in client.spaces.list() if space["name"] == "Work")

client.memories.remember(
    "We chose Postgres for the ledger, not DynamoDB.",
    space_ids=[work["id"]],
)
found = client.search.query("what did we decide about the ledger")
print([one["memory"]["title"] for one in found["results"]])

Three things to know first

remember does not return a memory. It hands material to the ingestion pipeline and answers 202 with a job id. Extraction, entity resolution, deduplication and conflict detection all run afterwards, and may produce one memory, several, or none. Poll client.jobs.get(job_id) if you need to know when - the terminal success state is completed, not succeeded.

A capture needs a Space. remember without space_ids is refused with a 400 that names the Spaces this account actually has. There used to be three fallbacks — an account default, the oldest Space, or a new one invented on the spot — and none of them exists now, because a memory filed somewhere nobody chose is a memory nobody finds again. client.spaces.working() answers which Space a given context is set to write to.

Search degrades rather than fails. With embeddings unavailable it falls back to deterministic retrieval and still answers. Read response["diagnostics"]["degraded"] before telling a user the system knows nothing; it may merely be looking with one eye.

POSTs are not retried unless you say they are safe. A POST that timed out may already have been processed - a connection that died after the server accepted the request is indistinguishable from one that died before. Pass an idempotency_key and the API deduplicates on it, so a retry returns the first result instead of doing the work twice:

client.memories.remember(
    "The migration is halfway done.",
    idempotency_key="standup:2026-08-28",
)

Give the key meaning. A fresh random value per call makes every retry a new request, which is exactly what it exists to prevent.

What is here

Every resource hangs off the client, and the async client mirrors it exactly — AsyncPersistMemory has the same attributes with the same methods, and a test fails the build if the two ever disagree.

A realistic example

Assembling context for a model, and filing what came back:

from persistmemory import AsyncPersistMemory, NotFoundError, RateLimitError


async def answer(question: str, conversation_id: str) -> str:
    async with AsyncPersistMemory(timeout=15.0, max_attempts=4) as client:
        # Bounded by TOKENS, not row count - ten long memories overflow a
        # window that fifty short ones fit inside.
        context = await client.search.context(
            question,
            token_budget=1_500,
            scope="combined",
            space_ids=["sp_work"],
        )
        if context["truncated"]:
            print("Something relevant was left out for budget.")

        reply = await call_your_model(context["context"], question)

        # Both turns in one call: a claim is often split across an exchange,
        # and extracting each turn in isolation finds neither half.
        appended = await client.conversations.append(
            conversation_id,
            [
                {"role": "user", "content": question},
                {"role": "assistant", "content": reply},
            ],
            idempotency_key=f"{conversation_id}:{question[:40]}",
        )
        if not appended["extracting"]:
            # Said out loud rather than assumed. With no queue configured the
            # turns are stored and never become memory.
            print(appended.get("note"))

        return reply

Paging, which every list endpoint shares:

# One page, for a UI that renders one page at a time.
first = client.memories.list(type="decision", limit=50).first()

# Or the whole walk. `all` needs a bound - an unbounded collect on a large
# account is minutes of requests and a list that exhausts memory.
decisions = client.memories.list(type="decision").all(500)

# Or item by item, stopping when you like.
for memory in client.memories.list(scope="combined"):
    if memory["confidence"] < 0.5:
        break

# The async client iterates the same way.
async for memory in client.memories.list(type="fact"):
    ...

Errors, which are classes you can branch on:

try:
    client.spaces.get("sp_missing")
except NotFoundError:
    return None
except RateLimitError as limited:
    # Already retried, and still refused. `retry_after_seconds` is what the
    # server said, not a guess.
    print(f"Rate limited; try again in {limited.retry_after_seconds or 60}s")

Branch on the class or on error.code, never on error.message. Messages get rewritten, translated, and deliberately made vaguer for security; a client keyed to message text breaks silently when any of that happens.

Sync and async

Both are written out, rather than one wrapping the other. A sync facade over the async client needs an event loop, and calling it from inside a running one either deadlocks or needs a background thread nobody asked for. What the two share is everything they DECIDE - headers, error mapping, retry policy, and the URL and body of every endpoint, in _core.py and _ops.py, which both import. The duplication is the waiting, and only the waiting.

Behaviour

The API key is held in a name-mangled attribute and never returned, logged or put in an error. repr(client) gives PersistMemory(base_url=..., api_key='[redacted]'), the Authorization header is built per request rather than stored on the httpx client where a debugger would print it, and any key-shaped string in an error message is redacted on the way out - because the way a credential actually escapes is a traceback pasted into a bug report, not a deliberate log line.

Options

from persistmemory import Backoff, PersistMemory

client = PersistMemory(
    api_key="pm_live_...",
    base_url="https://api.persistmemory.com",
    timeout=30.0,
    max_attempts=3,
    backoff=Backoff(base_seconds=0.25, max_seconds=8.0, factor=2.0, jitter=1.0),
    # Injected, which is what makes a test of your code incapable of opening a
    # socket by accident.
    transport=httpx.MockTransport(handler),
)

Tests

uv run pytest        # no network: every test runs against httpx.MockTransport
uv run ruff check .