Python SDK reference

Import

import ellivate

ellivate_client.py is written to your project root at publish time. It uses only Python stdlib — no pip dependencies. The ellivate object is a singleton.

KV storage — ellivate.get/set/del/list

get(key: str) -> Any

Read a value. Returns None if the key doesn't exist or the viewer lacks permission.

items = ellivate.get("shared:items")
if items is None:
    items = []

set(key: str, value: Any) -> None

Write a value. value is anything json.dumps-serializable. Raises on 401.

ellivate.set("shared:items", ["a", "b", "c"])
ellivate.set("personal:settings", {"theme": "dark"})

delete(key: str) -> None

Remove a key. Idempotent.

ellivate.delete("shared:items")

list(prefix: str) -> list[str]

Return all keys matching a prefix.

keys = ellivate.list("shared:tasks:")
# ["shared:tasks:abc", "shared:tasks:def"]

Collections — ellivate.collection(name)

The "many similar things" primitive — a sibling to KV. Use this when the shape is a list of items you'd want to filter, sort, or paginate (a roster, a feed, a list of expenses) rather than a single value. Each item is a dict with platform-managed id, createdAt, updatedAt fields.

collection(name: str) -> Collection

Open a handle for a collection. The collection is created on first add() — no separate provisioning step. Same scope rule as KV: the prefix on the name decides who sees it. No prefix or personal: is private per viewer (the default); ellivate.collection("shared:trips") is shared with the group; "public:trips" is readable by anyone. There is no scope argument — the prefix is it.

add(value: dict) -> dict

Add an item. The platform assigns id, createdAt, updatedAt and returns the full record.

things = ellivate.collection("things")          # private per viewer
trips = ellivate.collection("shared:trips")     # shared with the group

item = things.add({"name": "milk", "category": "dairy"})
# → {"id": "ck1...", "name": "milk", "category": "dairy",
#     "createdAt": "...", "updatedAt": "..."}

list(where: dict | None = None, sort: str = "-createdAt", limit: int = 100, cursor: str | None = None) -> dict

Return a page of items as {\"items\": [...], \"nextCursor\": ...}. Default sort is newest-first. Pass a cursor from a previous result to fetch the next page.

# Newest 100, no filter:
page = things.list()

# Filtered + paginated:
dairy = things.list(where={"category": "dairy", "in_stock": True}, limit=50)

# Next page:
if dairy["nextCursor"]:
    more = things.list(where={"category": "dairy"}, cursor=dairy["nextCursor"])

get(item_id: str) -> dict | None

Fetch a single item by id. Returns None for missing items.

update(item_id: str, patch: dict) -> dict

Shallow-merge a patch into an existing item. Untouched fields are preserved; updatedAt bumps.

things.update(item["id"], {"category": "dairy-and-cheese"})
# other fields untouched, updatedAt refreshed

replace(item_id: str, value: dict) -> dict

Replace an item's value entirely. id and createdAt are preserved; updatedAt bumps. Anything not in value is lost.

delete(item_id: str) -> None

Delete an item. Idempotent — deleting a missing item does not raise.

Blob storage — ellivate.blob

For binary data (photos, audio, documents). KV is JSON-only and capped at 256KB; blobs hold up to 25MB and serve from a CDN. See blob storage for the model.

put(data: bytes, mime_type: str = "application/octet-stream", scope: str | None = None) → dict

Upload bytes. Returns a dict with url, blobId, mimeType, sizeBytes, scope, createdAt.

with open(path, "rb") as f:
    result = ellivate.blob.put(f.read(), mime_type="image/jpeg")
ellivate.set("photos/" + slug, {"url": result["url"], "caption": caption})

get(url: str) → bytes | None

Fetch the bytes for a blob URL. Returns None on 404. Most apps don't need this — pass the URL through to a template / response and let the browser fetch.

list(scope: str | None = None) → list[dict]

List blobs in the current scope. Each entry has the same shape as put's return value.

delete(url: str) → None

Delete a blob. Idempotent on missing blobs. PERSONAL blobs can only be deleted by the original uploader; SHARED blobs by any group member.

Reasoning — ellivate.reason

Single-shot LLM completion, proxied server-side; the published app never sees a provider key. Billed to the caller — the viewer who triggered the call, not the app owner. The owner is charged only for their own usage and for owner-context calls with no viewer (scheduled / background fires). A per-app monthly soft cap still applies. See Pricing & credits for the caller-pays model.

reason(prompt: str, system: str | None = None, model: str | None = None, max_tokens: int | None = None, images: list | None = None, schema: dict | None = None, raw: bool = False) -> str | dict | list

Two modes, picked by whether you pass schema:

• No schema → returns a str, the model's text output.
• With schema (a JSON Schema dict) → returns the parsed, validated value (a dict or list) — index it directly, e.g. result["items"]. No json.loads.

Pass raw=True to get the full result dict with usage metadata (text in string mode, data in schema mode).

# Simple — returns a string.
summary = ellivate.reason(prompt=f"Summarize: {text}")

# Override the model when default isn't right.
draft = ellivate.reason(
    prompt=f"Draft a warm reply to: {thread}",
    system="You write warm, concise replies.",
    model="claude-sonnet-4-6",
    max_tokens=400,
)

# Structured output — pass a JSON Schema dict, get a parsed,
# validated value back.
result = ellivate.reason(
    prompt=f"Turn this grocery dump into a categorized list: {dump}",
    schema={
        "type": "object",
        "properties": {
            "items": {
                "type": "array",
                "items": {
                    "type": "object",
                    "properties": {
                        "category": {"type": "string"},
                        "item": {"type": "string"},
                    },
                    "required": ["category", "item"],
                },
            },
        },
        "required": ["items"],
    },
)
for entry in result["items"]:    # already parsed — no json.loads
    print(entry["category"], entry["item"])

# Full result dict via raw=True. "text" in string mode, "data" in
# schema mode.
full = ellivate.reason(prompt="...", raw=True)
# {
#   "text": str,            # string mode only
#   "data"?: ...,            # schema mode: the parsed value
#   "model": str,
#   "provider": "anthropic" | "openai" | "google",
#   "inputTokens": int,
#   "outputTokens": int,
#   "estimatedCents": int,
#   "monthToDate": {...},
#   "warning"?: "approaching_cap" | "exceeded_cap",
# }

# Attach images. URL form lets the provider fetch the bytes; base64
# form delivers them inline. Up to 20 per call. Supported media types:
# image/jpeg, image/png, image/gif, image/webp.
total = ellivate.reason(
    prompt="Return only the total amount as a number — no currency, no commas.",
    images=[{"url": receipt_url}],
)

caption = ellivate.reason(
    prompt="Write a short, friendly caption for this photo.",
    images=[{"base64": b64, "mediaType": "image/jpeg"}],
)

Images: attach pictures via images. Each entry is either {'url': '...'} (provider fetches the bytes) or {'base64': '...', 'mediaType': '...'} (pre-encoded). Up to 20 images per call. Google's Gemini accepts base64 only — use that form or pick a Claude / OpenAI model for URL images.

Structured output — pass a schema, get parsed data

When you need JSON back, don't ask for it in the prompt and parse the text — pass a schema (a JSON Schema dict). The SDK uses the provider's native structured-output mode (Anthropic tool-use, OpenAI json_schema, Gemini responseSchema), validates the result against your schema, and returns the parsed value. No fence-stripping, no regex extraction, no json.loads that raises on a stray prose sentence.

Wrong (asks for JSON in the prompt, parses the text — truncates silently when the output is long):

raw = ellivate.reason(
    prompt=f"Return JSON: {{ items: [...] }} for {dump}",
    max_tokens=2000,
)
items = json.loads(extract_json(raw))["items"]  # JSONDecodeError when cut off

Right (schema mode — validated, parsed):

result = ellivate.reason(
    prompt=f"Turn this grocery dump into a categorized list: {dump}",
    schema={
        "type": "object",
        "properties": {
            "items": {
                "type": "array",
                "items": {
                    "type": "object",
                    "properties": {
                        "category": {"type": "string"},
                        "item": {"type": "string"},
                    },
                    "required": ["category", "item"],
                },
            },
        },
        "required": ["items"],
    },
)
items = result["items"]   # already parsed

Token defaults + truncation

Output defaults are generous so a normal call doesn't get clipped: 4096 tokens for string mode, 8192–16384 for structured mode (a JSON object costs more tokens than the same answer in prose). Override with max_tokens when you know the shape is bigger.

If the model hits the cap before finishing, reason now raises a clear error rather than returning a half-formed answer. Structured mode: “output too large to finish — narrow the request or split it into smaller calls.” String mode: “output was cut off — raise max_tokens, narrow the request, or split it.” A silently truncated string that your code then tried to json.loads was the old failure mode; the raise makes it loud.

Out of credits: if the caller has no AI credits and no provider key, reason raises on an HTTP 402 AI_BUDGET_REQUIRED response — the inference runs on the caller's budget and there is no silent fallback to the owner's credits. Catch it to surface a friendly “you're out of AI credits” state instead of a broken screen.

Models:

Model name picks the provider. Omit model to let the platform pick the cheapest cheap-default across configured providers.

State-aware reason — wire your tool's data into the prompt

The single most common reason-related failure: a chat / coach / advisor surface ships, the user asks a question about their own data in the tool, and the model says “I can't see that.” The model only sees the prompt text — it has no idea what's in the tool's KV / Collections / blobs unless you put it there.

For any question whose answer depends on something stored in the tool, fetch the relevant state and pass it via the system prompt. For pure transformations (translate, format, generate from input) — skip the wiring; tool state isn't needed.

Wrong (state-aware question, no context):

def ask(user_message: str) -> str:
    return ellivate.reason(
        prompt=user_message,
        system="You are a lacrosse coach.",
    )

Right (fetch what the question implies, attach as context):

import json

def ask(user_message: str, today_id: str) -> str:
    today = ellivate.collection("sessions").get(today_id)
    drills = ellivate.collection("drills").list(
        where={"id": {"in": today["drillIds"]}}
    )
    system = f"""You are a lacrosse coach.

The athlete's session today ({today["date"]}):
{json.dumps(today, indent=2, default=str)}

Drills assigned:
{json.dumps(drills, indent=2, default=str)}

Answer using this context."""
    return ellivate.reason(prompt=user_message, system=system)

For multi-turn chat surfaces, re-fetch tool state on every turn. State changes between turns; cached context goes stale. The cost is small; the alternative is the bug.

The line: does the answer depend on something stored in the tool, or only on the prompt text? If only the prompt → skip the context wiring. If anything stored → fetch it and attach.

Time — ellivate.schedule

Scheduled HTTP callbacks. The platform calls back to the app at the scheduled time — the app's existing routes handle the fire. Same shape as the JS SDK with one Python quirk: the relative-delay verb is in_ (with trailing underscore) because in is a reserved keyword.

schedule.at(when: str, handler: str, data: dict | None = None)

One-time fire at a UTC ISO datetime.

r = ellivate.schedule.at(
    when="2026-12-25T08:00:00Z",
    handler="/api/birthday-reminder",
    data={"person": "Alice"},
)
# {"id": "ck1...", "nextFireAt": "2026-12-25T08:00:00.000Z"}

schedule.cron(cron: str, handler: str, data: dict | None = None, timezone: str | None = None)

Recurring 5-field cron. Evaluated in the optional timezone (IANA name); when omitted the platform falls back to the owner’s profile timezone (auto-detected from device) and then to UTC. Pass timezone whenever the user picked a wall-clock time. Min interval 1 minute.

ellivate.schedule.cron(
    cron="0 9 * * 1",                # every Monday at 9am in the named tz
    timezone="America/New_York",
    handler="/api/weekly-digest",
)

schedule.in_(handler: str, data: dict | None = None, *, seconds=None, minutes=None, hours=None, days=None)

Relative delay — sum of seconds / minutes / hours / days.

ellivate.schedule.in_(
    handler="/api/follow-up",
    days=7,
    data={"jobId": "abc"},
)

Read + cancel

page = ellivate.schedule.list(status="ACTIVE", limit=100)
# {"items": [...], "nextCursor": ...}

one = ellivate.schedule.get(schedule_id)
# dict or None

ellivate.schedule.cancel(schedule_id)

What the handler receives

The platform POSTs the handler with body { scheduleId, firedAt, data } and headersX-Ellivate-Schedule-Fire: true, X-Ellivate-App-Id, and (if scoped to a viewer) X-Ellivate-Viewer-Token. Return 2xx within 30s. 5xx / timeout retries up to 3 times with 1m / 5m / 30m backoff. 4xx is treated as the app's bug — no retry.

Notifications — ellivate.notify

Send a push notification to an Ellivate user. Delivery rides on the Expo push tokens the mobile shell already registered — your app never touches a native API. See Notifications for the full mental model.

notify(title, body, to=None, deep_link=None, data=None, card=None) → dict

Server-only and viewer-required. The call must run inside a request handler so Ellivate can identify the "from" via the per-request viewer token (set by ellivate-auth middleware). Cron jobs and startup code can't notify until the cron primitive ships its own service-account identity. Raises RuntimeError on missing config, missing viewer, or HTTP error.

from flask import Flask
import ellivate

app = Flask(__name__)

@app.route("/check")
def check():
    if found_a_match():
        ellivate.notify(
            to="owner",
            title="Match found",
            body="Open the app for details.",
            deep_link="/results",
        )
    return {"ok": True}

Targeting (to):

Posting to a Space — the card payload. When you target a Space ({"spaceId": ...} or {"space": "current"}), the notification renders as a card in the Space chat. Pass an optional card dict to give it structure — labeled fields, a tappable actionPath / actionLabel, and a thumbnailUrl. The card payload is ignored for non-Space targets.

# Post a card into the Space this tool was opened from.
ellivate.notify(
    title="Trip booked",
    body="Friday 7pm at Lazy Bear.",
    to={"space": "current"},
    card={
        "fields": [
            {"label": "When", "value": "Fri 7:00 PM"},
            {"label": "Party", "value": "4"},
        ],
        "actionPath": "/reservations/42",   # path inside the app
        "actionLabel": "View reservation",
        "thumbnailUrl": "https://example.com/lazy-bear.jpg",
    },
)

Returns: {"delivered": int, "skipped": int, "throttled": int} — each 0 or 1 per call.

Rate limits: 5/hour and 50/day per (app, sender, recipient) pair; 100/hour per app across all recipients. Over-quota raises RuntimeError with retryAfterSeconds in the message.

Persistence + badges. Every notify also persists a record that drives an in-app red bubble on the recipient's tile (mobile home grid + web dashboard) and the native app-icon badge count. Both clear when the recipient opens the tool — no extra call required. Because each fire is loud (push + badge), reserve notify for events the user should look at when they get a moment; for noisier state changes, write to KV or a Collection and let the user pull instead.

OAuth integrations — ellivate.connection(name)

Returns a typed connection handle whose .fetch() and .fetch_json() helpers attach a valid, auto-refreshing OAuth access token to every request. The user authorizes once in the Ellivate shell; every tool that uses the integration just works. See Connections & integrations for the full mental model.

connection(name: str, scope: str = "PERSONAL", oauth_scopes: list[str] | None = None)

Synchronous and cheap — returns a handle. Token-fetch + the provider API call happen on the first .fetch() / .fetch_json(). Raises ValueError immediately on an unknown integration name.

import ellivate

cal = ellivate.connection("google-calendar")

try:
    event = cal.fetch_json(
        cal.api_base + "/calendars/primary/events",
        method="POST",
        json_body={
            "summary": "Lunch",
            "start": {"dateTime": "2026-06-01T12:00:00-07:00"},
            "end":   {"dateTime": "2026-06-01T13:00:00-07:00"},
        },
    )
except ellivate.EllivateConnectionNeeded as e:
    # Render a "Connect your Google account" prompt
    return render_connect_prompt(e.provider_display_name)

Connection methods:

# Authenticated request — returns (status_code: int, body: bytes)
status, body = cal.fetch(url, method="GET", headers=None, data=None, json_body=None)

# Same plus JSON parse + RuntimeError on non-2xx
result = cal.fetch_json(url, method="GET", headers=None, data=None, json_body=None)

# Attributes
cal.integration         # "google-calendar"
cal.provider            # "google"
cal.scope               # "PERSONAL" | "SHARED"
cal.oauth_scopes        # ["https://www.googleapis.com/auth/calendar.events", ...]
cal.api_base            # "https://www.googleapis.com/calendar/v3"

EllivateConnectionNeeded

Raised on the first call when no usable connection exists. Catch it and render a "Connect [Provider]" prompt; the Ellivate shell handles the OAuth flow when the user taps.

class EllivateConnectionNeeded(Exception):
    integration: str
    provider: str | None
    provider_display_name: str | None
    scope: str  # "PERSONAL" | "SHARED"
    oauth_scopes: list[str]
    missing_scopes: list[str] | None  # set when existing connection is scope-deficient

PERSONAL vs SHARED

Defaults to PERSONAL — each viewer connects their own account. Pass scope="SHARED" only when the tool operates on the builder's account for every viewer (e.g. a family meal planner writing to the family calendar). SHARED-scoped connections block publish until the builder has connected; PERSONAL connections resolve lazily at runtime.

Identity — ellivate.viewer()

viewer = ellivate.viewer()
# { "id": "...", "email": "...", "username": "...", "verified": True } or None

Returns None for anonymous visitors on public + open apps. For authorization, trust that the middleware already gated out unauth'd visitors — use the viewer object for display only.

Scope prefixes

The key (or collection name) prefix decides who can see the data, resolved deterministically at runtime. There is no scope argument on kv or collection — the prefix is the whole mechanism. See Persistence & scopes for full details.

A key with no prefix is personal: — private to each viewer. There's no classifier guessing from the name; forget the prefix and your data stays private. Add shared: when you want the group to see it. blob.put and connection keep their explicit scope argument — those aren't prefix-driven.

Per-viewer scoping on Flask / FastAPI

When your Flask or FastAPI app runs under Ellivate, the auth middleware stashes the viewer token on a contextvars.ContextVar after verifying the cookie. The SDK reads that contextvar on every KV call and includes the viewer token in the request.

This means personal: and shared: keys are correctly scoped per viewer with no extra code — two concurrent users hitting ellivate.get("personal:draft") get their own drafts without you writing any request-plumbing yourself.

Server-lane fallback

For Streamlit, standalone scripts, and startup-time code (@app.on_event("startup") in FastAPI), there's no per-request viewer. The SDK falls back to server lane — reads and writes land in a single app-wide namespace.

This is usually what you want for single-user Python apps. If you meant per-user and you're getting cross-user data bleed, move the call inside a route handler.

Environment variables

Injected by Ellivate at deploy time:

• ELLIVATE_DATA_URL — KV backend URL.
• ELLIVATE_DATA_KEY — API key.
• ELLIVATE_APP_ID — your app's ID.
• ELLIVATE_VIEWER_TOKEN — server-lane fallback token (for scripts and startup code).

JSON types

Values are serialized via Python's built-in json module. That means:

• dict, list, str, int, float, bool, None — all work.
• datetime, Decimal, bytes, set, custom classes — convert to JSON-safe shapes before calling set.
• Dataclasses work via dataclasses.asdict(). Pydantic models via .model_dump().

Concurrency

Individual writes are atomic; there's no multi-key transaction primitive. For read-modify-write on shared collections, the SDK is safe to call but you're responsible for the logic — last write wins.

contextvars are async-safe (each asyncio task has its own context) and thread-safe (each thread, too). Concurrent requests on Flask's thread pool or FastAPI's event loop don't leak viewer context across each other.

What's next

• JavaScript SDK reference
• Persistence & scopes