> ## Documentation Index
> Fetch the complete documentation index at: https://docs.keeps.dev/llms.txt
> Use this file to discover all available pages before exploring further.
# Recording provider calls
> Instrument fal, Replicate, and OpenAI calls by hand — submit, run, finish — all on one asset_id.
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()`](/api-reference) 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](/api-reference).
## The pattern
Every generation is the same three-beat shape, no matter the provider:
Call `capture(model=..., inputs=..., status="submitted")` with no
`asset_id`. keeps mints a uuid7 and returns it. Hold onto that string.
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.
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](/privacy-and-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.
```python theme={null}
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:
```python theme={null}
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.
```python theme={null}
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.
```python theme={null}
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:
```python theme={null}
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.
```python theme={null}
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()`](/core-concepts) instead of repeating the literal — they're
merged into each `capture()` automatically (per-call values still win per key):
```python theme={null}
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](/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.
```python theme={null}
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"]}
```
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.
```python theme={null}
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](#example-b-an-async-polling-job):
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.