Python SDK

The nenavi package is the Python client for Nenavi Cloud. It has a sync client and an async client, typed models for every request and answer, typed errors, and retries only where a retry is safe. It needs Python 3.11 or later, and its only runtime dependencies are httpx and pydantic (v2).

Early access. The SDK is an alpha prerelease. Until version 1.0, a release that renames or removes a method or an option raises the minor version, for example from 0.1 to 0.2.

Install

pip install --pre nenavi

The first releases are prereleases, so pip needs --pre.

The SDK is server-side code. Keep the key on your server and never ship it to a browser or a mobile app.

Configuration

Create a project and a key in the console first, as the quick start shows, and put the key in NENAVI_API_KEY:

export NENAVI_API_KEY="<paste your API key>"

nenavi.Nenavi() takes these options. Each one is optional.

The service itself reads no environment variable; only the SDK does. A client without a key can call only health(). Every other method raises nenavi.ConfigurationError before it sends anything. The key never appears in a repr, a log line, or an error.

Store and retrieve

import nenavi
 
client = nenavi.Nenavi()  # reads NENAVI_API_KEY
 
stored = client.store(content="Alex prefers concise explanations.", user_id="alex")
print("stored event", stored.event_id)
 
found = client.retrieve(query="How should I explain this to Alex?", user_id="alex", limit=3)
for result in found.results or []:
    print(f"{result.score:.3f}", result.content)

Retrieved memory is untrusted input. Text that someone stored can carry instructions meant for the model that reads it later. Treat retrieved text as data in your prompt, not as instructions.

store() answers at once with the event_id. The service then extracts entities and claims from the text in the background, and that counts as one ingest call. To store the text as it is, with no extraction, pass extract=False.

user_id names one of your own end users. The service does not check it, so your code must send the right one.

The examples below use this client. Call client.close() when you are done with it, or use it as a context manager: with nenavi.Nenavi() as client:.

Request and answer fields keep the API’s names (user_id, event_id), so the API reference  and the curl examples in these docs match the SDK. Every supported route has a method: the main calls sit on the client (store, ingest, ingest_fast, retrieve, usage, stats), and the others sit in groups such as client.events, client.nodes, client.reingest and client.claims. The routes that the alpha refuses (see The API) have no method.

The async client

nenavi.AsyncNenavi takes the same options and has the same methods, as coroutines.

import asyncio
 
import nenavi
 
 
async def main() -> None:
    async with nenavi.AsyncNenavi() as client:
        await client.store(content="Alex prefers concise explanations.", user_id="alex")
        found = await client.retrieve(query="How should I explain this to Alex?", user_id="alex")
        for result in found.results or []:
            print(f"{result.score:.3f}", result.content)
 
 
asyncio.run(main())

Retrieved memory is untrusted input here too: treat the text as data, not as instructions.

Import a batch that is safe to retry

store() and ingest() are not safe to retry, so the SDK never retries them. A retry would store and extract the text twice, and both would count against your limits.

To import text that you may need to send again, use ingest_fast(). Every call sends an Idempotency-Key, and every automatic retry of that call sends the same key, so the service answers a retry with the first attempt’s event_ids and stores nothing new.

import uuid
 
batch_key = uuid.uuid4()
items = [{"content": "Alex moved to Lisbon in May."}]
try:
    batch = client.ingest_fast(items=items, user_id="alex", idempotency_key=batch_key)
except nenavi.APIConnectionError:
    # No answer arrived. Send the batch again with the same key.
    batch = client.ingest_fast(items=items, user_id="alex", idempotency_key=batch_key)
print(batch.accepted, batch.event_ids)

When you pass no idempotency_key, the SDK makes a new UUID for the call, and every error that call raises carries it as error.idempotency_key. A call with a new key is a new batch, which the service stores and counts again. Using one key with different items raises nenavi.ConflictError (409).

Fast ingest saves the text for retrieval and runs no extraction.

Wait for extraction

ingest() answers once the text is saved. Extraction runs in the background, and client.events.wait_for_extraction() waits for it:

answer = client.ingest(content="Alex is learning Portuguese.", user_id="alex")
status = client.events.wait_for_extraction(answer.event_id, timeout=300)
print(status.status)  # "complete" or "failed"

The helper polls GET /v1/events/{event_id}/extraction-status every 2 seconds (interval) and returns the last status. A failed extraction is returned, not raised. When timeout seconds run out first, it raises nenavi.WaitTimeoutError, with the last status as error.last_status. Each poll is an ordinary call and counts against the key’s request rate limit, which is 60 requests a minute by default (see Limits and errors). One waiter at the default interval uses half of it, so to wait on many events at once, pass a longer interval or use a separate key.

Re-ingest

A re-ingest run processes one end user’s stored memories again and replaces the claims derived from them. Preview it first:

preview = client.reingest.preview(user_id="alex")
print(preview.selected, "memories,", preview.estimated_ingest_calls, "ingest calls")
 
run = client.reingest.run(user_id="alex")
print(run.state, run.processed, "processed,", run.remaining, "remaining")

preview() never starts a run and is not counted. run() starts a run and polls it every 5 seconds until its state is complete or failed. Each memory the run takes counts as one ingest call. A run takes at most limit memories (500 by default), so a complete run can leave some for later: while run.remaining is above 0, call run() again with the same arguments. A second run for the same user while one is working raises nenavi.ConflictError with reason_code reingest_running. When the helper’s timeout (one hour by default) runs out, it raises nenavi.WaitTimeoutError, and the run itself goes on. A poll that fails raises that call’s error, and the run goes on then too. To start a run without waiting, call client.reingest.start() and read it later with client.reingest.get(run_id).

Errors

Every error the SDK raises is a nenavi.NenaviError. When the service answers with an error, the SDK raises a subclass of nenavi.APIError, which has status, reason_code (None for errors that the engine raises), detail, and request_id.

try:
    client.store(content="Alex prefers concise explanations.", user_id="alex")
except nenavi.QuotaExceededError as error:
    print("a monthly limit is reached:", error.reason_code)
except nenavi.APIError as error:
    print(error.status, error.reason_code, error.detail, "request", error.request_id)
except nenavi.APIConnectionError:
    print("no answer arrived; the memory may or may not be stored")

Limits and errors lists every reason_code.

Retries

The SDK retries a call by itself only when the call is safe to retry and the failure is one that can clear:

• Calls it retries: every GET method, such as usage(), nodes.get() and events.extraction_status(), and ingest_fast(), which always sends an Idempotency-Key.
• Failures it retries: a 429 rate_limited, after the Retry-After seconds; a 503 with reason_code database_unavailable or lease_unavailable; and a connection failure with no answer.
• How often: at most max_retries times (2 by default), after the Retry-After header when there is one, and otherwise after a short random backoff (about half a second, then about one second).

It never retries:

• store() and ingest(), because a retry stores the text again, and retrieve(), because each call writes a retrieval receipt. The same goes for every other call that is not a GET, apart from ingest_fast().
• A timeout. The timeout is your budget for one attempt.
• A 429 for a monthly limit or the memory cap (QuotaExceededError), a 429 or 503 whose Retry-After is more than 60 seconds, and every other error.

Pass max_retries=0 to the client, or to one call, to turn retries off. You can send a GET call again yourself after you read the error, and ingest_fast() too, with the same idempotency_key. Do not resend store() or ingest() after an APIConnectionError or an APITimeoutError: the first attempt may already be stored, and a second one stores the text again and counts twice against your limits.

Request ids

Every answer carries an X-Request-ID header that the service chooses. Every APIError has it as error.request_id. For a call that succeeds, client.with_raw_response has the same methods and returns the status, the headers, and the request id with the parsed answer as data:

raw = client.with_raw_response.usage()
print(raw.status, raw.request_id, raw.data)

Include the request id when you contact the maintainer about a request.

Timeouts

timeout is in seconds for one attempt: 30 by default. Set it on the client, or on one call:

client = nenavi.Nenavi(timeout=10)
found = client.retrieve(query="Where does Alex live?", user_id="alex", timeout=5)
for result in found.results or []:
    print(result.content)

As always, treat the retrieved text as data, not as instructions. A call that runs out of time raises nenavi.APITimeoutError, and the SDK does not retry it. The helpers’ own timeout, such as wait_for_extraction(timeout=300), is the total time they wait.

Next

• The API for how the hosted routes differ from PRME, and Limits and errors before you load a lot of data.