API reference

The keeps SDK exposes exactly seven public functions. This page documents each one precisely: its signature, every parameter, what it returns, and a runnable example. The order below follows the lifecycle of a typical integration: init, capture, context, flush, session, shutdown, stats.

Environment variables. These are read by init when the matching argument is not passed explicitly:

KEEPS_API_KEY — your ingest API key. project_id is derived server-side from this key; the client never sends it.
KEEPS_INGEST_URL — overrides the events endpoint (staging / self-run). If unset, the hosted default https://ingest.keeps.dev/v1/events is used.
KEEPS_DISABLED — set to 1, true, or yes to disable the SDK entirely (handy in CI). Equivalent to init(disabled=True).

Legacy LENS_* env var names are still honored (KEEPS_* wins when both are set), and legacy lk_* API keys keep working forever — auth is by hash.

init

Configure the process-global client. Call once at startup. Fail-open: this never raises into your code — a missing API key logs a single warning and disables the SDK instead of erroring.

keeps.init(
    api_key=None,
    endpoint=None,
    *,
    user_id=None,
    disabled=None,
    flush_interval=1.0,
    max_buffer=10_000,
    transport=None,
) -> None

api_key

str | None

default:"None"

Your ingest API key. Falls back to the KEEPS_API_KEY environment variable. If still empty (and not disabled), the SDK logs a warning and disables itself.

endpoint

str | None

default:"None"

The events endpoint URL. Falls back to KEEPS_INGEST_URL. init auto-appends /v1/events if the URL lacks that path. When unset, the hosted default is used (https://ingest.keeps.dev/v1/events).

user_id

str | None

default:"None"

A default user_id applied to every capture() that does not supply one explicitly or via context().

disabled

bool | None

default:"None"

Force the SDK on or off. When None, the value is taken from KEEPS_DISABLED (1/true/yes → disabled). A disabled SDK still mints and returns asset_ids from capture() so your correlation logic keeps working.

flush_interval

float

default:"1.0"

Seconds between automatic background flushes of the in-memory buffer. The buffer also flushes immediately once 200 events accumulate.

max_buffer

int

default:"10_000"

Maximum number of events held in memory. On overflow the newest event is dropped and stats()['dropped_overflow'] increments.

transport

Transport | None

default:"None"

Inject a custom transport, primarily for testing. Most callers leave this unset and the default HTTP transport is used.

Returns: None.

import keeps

# Reads KEEPS_API_KEY / KEEPS_INGEST_URL from the environment if omitted.
keeps.init(api_key="kp_live_…", endpoint="https://ingest.example.com")

capture

Record one event about an asset and return its asset_id. This is a non-blocking O(1) append to an in-memory buffer — it returns immediately and never raises into your code. The first call (with asset_id=None) mints a uuid7 asset_id and returns it. Pass that id back on later calls to attach more events to the same generation (start → finish → annotate). Reusing asset_id is how correlation works — the backend groups by it.

keeps.capture(
    asset_id=None,
    *,
    user_id=None,
    model=None,
    provider=None,
    status=None,
    inputs=None,
    outputs=None,
    error=None,
    usage=None,
    properties=None,
    ts=None,
) -> str

asset_id

str | None

default:"None"

The id of the asset this event belongs to. None mints a fresh uuid7 and returns it; pass a previously-returned id to add to that same asset.

user_id

str | None

default:"None"

The end user this generation is for. Falls back to the ambient context() value, then the init(user_id=...) default.

model

str | None

default:"None"

The bare provider-native model id, e.g. "fal-ai/flux-pro/v1.1". Don’t prefix the vendor into it — that’s what provider is for.

provider

str | None

default:"None"

The vendor slug, e.g. "fal", "replicate", "openai" (lowercased server-side). A dimension for pricing and reliability breakdowns — the same model exists on multiple providers at different prices. It is never a correlation key: correlation is only the asset_id.

status

str | None

default:"None"

The lifecycle status, e.g. "submitted", "completed", "failed". Ingestion is out-of-order tolerant: the latest terminal status wins, and once an asset is terminal (completed/failed/cancelled) a non-terminal event can’t reopen it.

inputs

Any

default:"None"

The request inputs (typically a dict, e.g. {"prompt": "..."}).

outputs

Any

default:"None"

Output references — URLs and metadata. Typically a list of dicts, e.g. [{"url": "https://…"}].

error

dict[str, Any] | None

default:"None"

A verbatim provider error payload. The SDK ships it as-is; error normalization happens server-side.

usage

dict[str, Any] | None

default:"None"

The provider’s billing meters, verbatim, as a flat numeric map — e.g. {"duration_s": 6.2, "n": 1}. These numbers exist only in the provider’s response at call time and are unrecoverable later, so capture them here. Non-numeric values are dropped per key; an empty result is omitted.

properties

dict[str, Any] | None

default:"None"

Group-by tags stored server-side as Map(String, String). The fold is per key — for each key, the value from the latest event (by ts) wins — so you can append a partial bag without clobbering keys set on earlier events. Omitting properties never overwrites anything.Values get one canonical string encoding server-side: send True/False and they are stored as 'true'/'false'; numbers and containers are stored as compact JSON. A None value is a tombstone — it clears that tag. A bag over its 8192-byte budget has its values trimmed to fit (never replaced with a stub key), and the cut is flagged in the event’s truncated field.

str | None

default:"None"

An explicit event timestamp. When omitted the SDK stamps the event itself.

Returns: str — the asset_id for this event (always returned, even when the SDK is disabled or uninitialized).

import keeps

# First call mints and returns the id.
asset_id = keeps.capture(
    model="fal-ai/flux-pro/v1.1",
    provider="fal",
    status="submitted",
    inputs={"prompt": "a red bicycle"},
)

# Later, reuse the same id to attach the result.
keeps.capture(
    asset_id,
    status="completed",
    outputs=[{"url": "https://cdn.example.com/out.png"}],
    usage={"n": 1, "duration_s": 6.2},   # the provider's billing meters, verbatim
)

In short-lived, serverless, or forked workers, call flush() before the unit of work returns — otherwise buffered events can be lost when the process is frozen or killed before the background flush tick. See Reliability.

context

A context manager that tags every capture() inside the block with ambient user_id and arbitrary properties. Blocks nest (inner wins per key), follow the current thread/task, and propagate into asyncio tasks created inside the block.

keeps.context(*, user_id=None, **properties) -> context manager

user_id

str | None

default:"None"

Ambient user_id applied to every capture() in the block that does not pass its own user_id.

**properties

Any

Any remaining keyword arguments become ambient properties tags merged into each capture() in the block. Per-call properties win per key.

Returns: a context manager. It only tags — it does not flush on exit. Fail-open: a broken context never raises.

import keeps

with keeps.context(user_id="u_42", experiment_id="exp-123", tier="pro"):
    # This capture is tagged with user_id=u_42 and the two properties.
    keeps.capture(model="flux-pro", inputs={"prompt": "a tabby cat"})

flush

Block until buffered events are delivered (or dropped), up to timeout seconds. Call this before a short-lived process exits to avoid losing in-buffer events.

keeps.flush(timeout=5.0) -> bool

timeout

float | None

default:"5.0"

Maximum seconds to wait for the buffer to drain. None waits without a deadline.

Returns: bool — True if the buffer drained within the timeout, False if it timed out. (Also returns True immediately if the SDK was never initialized.)

import keeps

keeps.capture(model="flux-pro", inputs={"prompt": "a lighthouse"}, status="completed")

# Make sure it's delivered before the worker returns.
if not keeps.flush(timeout=3.0):
    # best-effort: log and move on, never block customer code
    ...

session

A context manager for one unit of work: it applies ambient tags like context and guarantees a flush on exit — the built-in form of the flush-before-exit pattern for serverless, worker, and agentic code.

keeps.session(*, user_id=None, flush_timeout=5.0, **properties)  # context manager

user_id

str | None

default:"None"

Ambient user_id applied to every capture() in the block (same resolution as context).

flush_timeout

float | None

default:"5.0"

Seconds to wait for the exit-flush to drain. None waits without a deadline.

**properties

Any

Remaining keyword arguments become ambient properties tags for the block.

Behavior: flushes on exit whether the block returns or raises. Fail-open — it never raises on its own and does not swallow exceptions from inside the block (it flushes first, then they propagate).

import keeps

def handler(event):
    with keeps.session(user_id=event["user"], job="render"):
        aid = keeps.capture(model="flux-pro", status="submitted",
                           inputs={"prompt": event["prompt"]})
        keeps.capture(aid, status="completed", outputs=[{"url": "https://…"}])
    # flushed here, before the function returns

shutdown

Flush and then close the transport, releasing the background flush thread. Call this on a clean application exit. An atexit handler also runs this on normal interpreter exit, so explicit calls are mainly for orderly teardown.

keeps.shutdown(timeout=2.0) -> None

timeout

float

default:"2.0"

Maximum seconds to wait while flushing before closing the transport.

Returns: None. A no-op if the SDK was never initialized.

import keeps

# At the end of a long-running process.
keeps.shutdown(timeout=5.0)

stats

Return a snapshot of SDK counters. Useful for monitoring delivery health and diagnosing dropped events. Counters are cumulative for the life of the process.

keeps.stats() -> dict

Returns: dict. Before init() it contains only {"initialized": False}. After init() it contains every key below:

initialized

bool

True once init() has run. When False, no other keys are present.

disabled

bool

True if the SDK is disabled (no API key, KEEPS_DISABLED, or disabled=True). Disabled clients still mint asset_ids but send nothing.

emitted

int

Total events accepted into the buffer by capture(). Your offered load.

sent

int

Total events successfully delivered to the ingest endpoint.

accepted

int

Total events the server confirmed it stored, parsed from each 202 response body. This is the server’s own count — the strongest delivery signal the SDK can give you.

rejected_by_server

int

Events the server rejected individually in the 202 body (bad timestamp, over-cap field, …). These were delivered but not stored, and retrying them cannot succeed. The first rejection also logs a one-time warning with the reason codes.

batches

int

Number of HTTP batches successfully POSTed. Events are batched up to 200 per request.

retries

int

Number of delivery retry attempts (exponential backoff, honoring 429 Retry-After). A high value signals a flaky or throttling endpoint.

dropped_overflow

int

Events dropped because the buffer was full (max_buffer). The newest event is dropped on overflow. A nonzero value means you are producing faster than the SDK can deliver — raise max_buffer or check connectivity.

dropped_send

int

Events dropped after delivery failed past the retry limit (or a non-retryable 4xx). A nonzero value means events were permanently lost in transit.

import keeps

keeps.flush()
print(keeps.stats())
# {'initialized': True, 'disabled': False, 'emitted': 12, 'sent': 12,
#  'batches': 1, 'retries': 0, 'dropped_overflow': 0, 'dropped_send': 0,
#  'accepted': 12, 'rejected_by_server': 0}

Delivery is best-effort, not at-least-once. emitted - sent (minus what is still buffered) tells you how many events the SDK could not deliver; the two dropped_* counters tell you why. sent counts what reached the server; accepted is what the server actually stored — a gap between them shows up in rejected_by_server. See Reliability.

​init

​capture

​context

​flush

​session

​shutdown

​stats

init

capture

context

flush

session

shutdown

stats