Skip to main content
There are no wrappers, no decorators, and nothing that patches fal_client, replicate, or openai. keeps never sits in your request path. You instrument a provider call by calling capture() around it yourself — once when you submit, once when it finishes.
If you came here looking for keeps.instrument(), an auto-patching hook, or a with keeps.generation(...) span — none of those exist. The entire surface is init, capture, context, session, flush, shutdown, stats. Manual capture() is the integration. See the API reference.

​
The pattern

Every generation is the same three-beat shape, no matter the provider:
1

Capture the submission

Call capture(model=..., inputs=..., status="submitted") with no asset_id. keeps mints a uuid7 and returns it. Hold onto that string.
2

Run your real provider call — untouched

Call fal / Replicate / OpenAI exactly as you would without keeps. Same arguments, same return value, same exceptions. keeps is side-band: it never sees or alters this call.
3

Capture the outcome on the same asset_id

Call capture(asset_id, status="completed", outputs=[...]) on success, or capture(asset_id, status="failed", error=...) on failure. Reusing the asset_id is what ties the two events to one generation server-side.
Reuse the same asset_id for every event about one generation. The first capture() mints it; pass it back on every later call. Correlation is GROUP BY asset_id on the backend — a fresh id means a fresh, orphaned generation. Nothing else correlates: provider= is a dimension (it powers pricing and per-vendor reliability breakdowns), never a join key, and the provider’s request id doesn’t thread events together either. Tag that request id as the reserved generation_id property — it’s the join key for reconciling against the provider’s invoice — but carry the asset_id yourself.

​
Side-band, always

Write the wrapping so that if you deleted the two capture() lines, your provider code would be byte-for-byte identical. keeps captures references to your inputs and outputs — it never changes the arguments you pass the provider, the value the provider returns, or the exception it raises. outputs are URLs and metadata only, never raw bytes (see Privacy & safety).

​
Example A — a synchronous call

A blocking fal/OpenAI-style call where submission and completion live in the same function. Mint on the way in, finish on the way out. 
import keeps
import fal_client

keeps.init()  # reads KEEPS_API_KEY / KEEPS_INGEST_URL from the environment

def generate_image(prompt: str, user_id: str) -> str:
    inputs = {"prompt": prompt, "image_size": "landscape_16_9"}

    # 1. Submit: mint the asset_id. `provider` is a breakdown dimension —
    #    it never correlates anything.
    asset_id = keeps.capture(
        model="fal-ai/flux-pro/v1.1",
        provider="fal",
        status="submitted",
        inputs=inputs,
        user_id=user_id,
    )

    # 2. The real provider call — exactly as you'd write it without keeps.
    result = fal_client.subscribe("fal-ai/flux-pro/v1.1", arguments=inputs)

    # 3. Finish on the SAME asset_id, outputs as references (URLs only).
    keeps.capture(
        asset_id,
        status="completed",
        outputs=[{"url": img["url"]} for img in result["images"]],
    )

    return result["images"][0]["url"]
The OpenAI shape is identical — only the call in the middle and the output shaping change: 
from openai import OpenAI

client = OpenAI()

def make_image(prompt: str) -> str:
    inputs = {"model": "gpt-image-1", "prompt": prompt, "size": "1024x1024"}

    asset_id = keeps.capture(model=inputs["model"], provider="openai",
                            status="submitted", inputs=inputs)

    resp = client.images.generate(**inputs)  # untouched

    keeps.capture(
        asset_id,
        status="completed",
        outputs=[{"url": d.url} for d in resp.data],  # references, not the image
    )
    return resp.data[0].url

​
Example B — an async / polling job

When submission and completion happen in different places — an HTTP handler submits, a worker or webhook finishes — the only thing you carry between them is the asset_id. Store it next to the provider’s request id (in your DB, your queue payload, wherever you already track the job) and pass it back. 
import keeps
import fal_client

# --- Process 1: the API handler submits the job ---
def submit_job(prompt: str, user_id: str) -> dict:
    inputs = {"prompt": prompt}

    asset_id = keeps.capture(
        model="fal-ai/flux-pro/v1.1",
        provider="fal",
        status="submitted",
        inputs=inputs,
        user_id=user_id,
    )

    handle = fal_client.submit("fal-ai/flux-pro/v1.1", arguments=inputs)

    # Tag the provider's request id as the reserved `generation_id` property —
    # the join key for reconciling against the provider's invoice later. A
    # properties-only event like this merges into the asset without touching
    # its status.
    keeps.capture(asset_id, properties={"generation_id": handle.request_id})

    # Persist OUR asset_id alongside the provider's request id so the worker
    # can pick the same thread back up later.
    save_job(request_id=handle.request_id, asset_id=asset_id)
    return {"request_id": handle.request_id}


# --- Process 2: the worker (or webhook) finishes the job ---
def finish_job(request_id: str) -> None:
    job = load_job(request_id)          # whatever you saved above
    asset_id = job["asset_id"]          # carry the SAME id across the boundary

    result = fal_client.result("fal-ai/flux-pro/v1.1", request_id)

    keeps.capture(
        asset_id,                       # reused → same generation server-side
        status="completed",
        outputs=[{"url": img["url"]} for img in result["images"]],
    )
    keeps.flush()                        # short-lived worker: deliver before exit
Out-of-order is fine. If the completion event lands before the submission event, ingestion still merges them: the fold orders events by their ts, not by arrival time, and a late-arriving submitted can never flip a finished asset back to in-flight. You don’t have to guarantee ordering — only that both events share the asset_id.

​
Example C — the failure path

Wrap the real call so its exception propagates unchanged, and record the provider’s error verbatim before re-raising. keeps ships errors as-is — normalization into a taxonomy happens server-side, so don’t pre-massage them. 
import keeps
import fal_client

def generate_or_raise(prompt: str, user_id: str) -> str:
    inputs = {"prompt": prompt}
    asset_id = keeps.capture(
        model="fal-ai/flux-pro/v1.1",
        status="submitted",
        inputs=inputs,
        user_id=user_id,
    )

    try:
        result = fal_client.subscribe("fal-ai/flux-pro/v1.1", arguments=inputs)
    except Exception as exc:
        # Record the failure on the SAME asset_id, error captured verbatim.
        keeps.capture(
            asset_id,
            status="failed",
            error={"type": type(exc).__name__, "message": str(exc)},
        )
        raise  # the caller sees the original exception — keeps never swallows it

    keeps.capture(
        asset_id,
        status="completed",
        outputs=[{"url": img["url"]} for img in result["images"]],
    )
    return result["images"][0]["url"]
If a provider returns a structured error body instead of raising (some HTTP clients do), pass that body straight through — keep the provider’s own code/message/status fields. error accepts a plain dict: 
resp = client.images.generate(**inputs)
if getattr(resp, "error", None):
    keeps.capture(asset_id, status="failed", error=dict(resp.error))  # verbatim
Don’t rewrite, classify, or collapse provider errors before capturing them. The raw message is the signal — server-side enrichment derives the error category from it. Reshaping it client-side throws that signal away.

​
Attaching your own dimensions

properties are GROUP BY tags (stored server-side as a string map) — film_id, shot_id, step, anything you want to break generations down by later. Two things to remember: they’re tags, not schema columns, and the server-side fold is per key — each capture() updates only the keys it carries (latest value per key wins), so you can append a partial bag on a later event and the keys you set earlier stay intact. 
tags = {"film_id": "f_88", "shot_id": "s_12", "step": "keyframe"}

asset_id = keeps.capture(
    model="fal-ai/flux-pro/v1.1",
    status="submitted",
    inputs=inputs,
    properties=tags,            # your dimensions
)

result = fal_client.subscribe("fal-ai/flux-pro/v1.1", arguments=inputs)

keeps.capture(
    asset_id,
    status="completed",
    outputs=[{"url": img["url"]} for img in result["images"]],
    properties=tags,            # tags from submit persist (per-key merge); resend optional
)
If the same tags apply to every call in a block, hoist them into keeps.context() instead of repeating the literal — they’re merged into each capture() automatically (per-call values still win per key): 
with keeps.context(user_id=user.id, film_id="f_88", shot_id="s_12"):
    asset_id = keeps.capture(model="fal-ai/flux-pro/v1.1",
                            status="submitted", inputs=inputs)
    result = fal_client.subscribe("fal-ai/flux-pro/v1.1", arguments=inputs)
    keeps.capture(asset_id, status="completed",
                 outputs=[{"url": img["url"]} for img in result["images"]])
Note context() only tags — it does not flush. See Reliability.

​
Serverless & short-lived workers: flush

capture() is a non-blocking append to an in-memory buffer; a daemon thread delivers it about once a second. If your process can be frozen or killed before that tick — a serverless function, a queue worker that exits, a forked job — call keeps.flush() before the unit of work returns, or the buffered (and in-flight) events are lost. Delivery is best-effort, not at-least-once. 
def handler(event, ctx):
    asset_id = keeps.capture(model="fal-ai/flux-pro/v1.1",
                            status="submitted", inputs=event["inputs"])
    result = fal_client.subscribe("fal-ai/flux-pro/v1.1",
                                  arguments=event["inputs"])
    keeps.capture(asset_id, status="completed",
                 outputs=[{"url": img["url"]} for img in result["images"]])

    keeps.flush()   # block until drained — the process is about to freeze
    return {"url": result["images"][0]["url"]}

Reliability model

Buffering, the flush daemon, overflow drops, retry/backoff, and exactly when you need an explicit flush() — all in one place.

​
Replicate, in one block

Same three beats. The model ref goes in model; the input map goes in inputs; outputs come back as URLs. 
import keeps
import replicate

def upscale(image_url: str, user_id: str) -> str:
    ref = "nightmareai/real-esrgan"
    inputs = {"image": image_url, "scale": 4}

    asset_id = keeps.capture(model=ref, provider="replicate",
                            status="submitted", inputs=inputs, user_id=user_id)

    output = replicate.run(ref, input=inputs)   # untouched

    keeps.capture(asset_id, status="completed", outputs=[{"url": str(output)}])
    return str(output)
For a split create/poll flow with replicate.predictions.create(...) then .get(id) in another process, follow Example B: persist the asset_id next to the prediction id and reuse it when you finish.

​
Checklist

Let the first capture() mint it; pass it back on every later call. Never mint a second id for the same generation.
No keeps argument changes what you send the provider or what it returns. Removing the capture() lines must leave provider behavior identical.
URLs and metadata only — never raw image/audio/video bytes.
Capture the provider’s error as-is and re-raise the original exception. Normalization is server-side.
Each capture() updates only the keys it carries (latest per key wins), so append partial bags freely — earlier keys persist.
Serverless / workers / forks: keeps.flush() before returning.