riffsdk

Python SDK for Riff. Provides sync and async clients for the Storage API, and lookups against an app's task board.

Installation

uv add riffsdk
# or
pip install riffsdk

Or install from a branch (for pre-release testing):

uv add git+https://github.com/databutton/riff-sdk-python.git@main
# or
pip install git+https://github.com/databutton/riff-sdk-python.git@main

Quick start

from riffsdk.storage import StorageClient, project_scope

client = StorageClient(scope=project_scope())

# Upload (content type derived from the key: text/plain; charset=utf-8)
meta = client.put("hello.txt", "Hello, world!")
meta = client.put("notes.md", "# Title")      # -> text/markdown; charset=utf-8

# Download
data = client.get("hello.txt")

# List
for obj in client.list("hello"):
    print(f"{obj.key} ({obj.size} bytes)")

# Delete
client.delete("hello.txt")

client.close()

Async

from riffsdk.storage import AsyncStorageClient, project_scope

async with AsyncStorageClient(scope=project_scope()) as client:
    await client.put("key", b"data", content_type="application/octet-stream")
    data = await client.get("key")

Tasks

Look one task up on the app's task board, in whatever state it is in — completed ones included:

import riffsdk.tasks.v0 as tasks

task = tasks.get_task("TASK-42")   # display ID as it appears on the board, or the task's id

if task is None:
    ...                            # nothing on the board for this item
elif task["status"] == "completed":
    ...                            # already handled

This exists for trigger tools. A trigger is handed the tasks that still have work left on them and nothing about the ones already finished, so it cannot tell an item it has never seen from one whose task is done — and proposes the same finished work every time it fires. get_task closes that gap.

Import the version you are writing against, bound to a short name as above. A version does not change under you: when the API changes it appears as a new version, and the one you imported keeps behaving as it did. Importing riffsdk or riffsdk.tasks loads no version.

API

• get_task(ref, *, client=None) -- returns a Task, or None if there is no such task. get_task_async is the async form.
• Task is the task as JSON plus the version of that shape. Read fields by name (task["status"], task["metadata"], task.get("summary"), or task.data for the whole dict), so a task gaining a field needs no SDK release.
• Raises TaskError if the lookup failed -- including on a persistent 429/5xx after retrying with backoff. A task that does not exist is not a failure; it comes back as None.
• client lets many calls (e.g. in a loop) share one connection. It defaults to a shared riffsdk.client.RiffClient/AsyncRiffClient built from the ambient environment the first time it's needed -- pass your own to point it elsewhere or to share it explicitly with other riffsdk.tasks/riffsdk.ai calls.
• The board read is the app's own: a deployed app sees its production tasks, the same app in the workspace sees the workspace's. There is nothing to configure for that, and no way to read the other one.
• Runs from an app's backend, where the environment it needs is already set.

AI

Extract text from a stored document -- a PDF, an image, anything that is not already plain text -- via an LLM, from an app's backend or from sandboxed generated code, without going through an agent session:

import riffsdk.ai.v0 as ai
from riffsdk.storage import project_scope

text = ai.interpret_document("invoice.pdf", scope=project_scope())

This exists because reading a PDF or an image otherwise requires the in-session interpret_document tool, which only an agent session has a handle on. scope is required -- pass project_scope(), account_scope(), or session_scope() to say which storage this reads from. prompt steers the extraction (e.g. "the line items only") when the default is not what you need. This is read-only: the result is returned, never written back to storage. riffsdk.ai is a separate module from riffsdk.tasks/ riffsdk.storage on purpose -- everything under it costs an LLM call, so it stays visually distinct from a plain storage read at the call site. To interpret a file you just uploaded, call it with the same key/scope you passed to riffsdk.storage's put/upload_file -- storage's client is unrelated to this module's, so there is nothing to share beyond the key.

Import the version you are writing against, bound to a short name as above. A version does not change under you: when the API changes it appears as a new version, and the one you imported keeps behaving as it did. Importing riffsdk or riffsdk.ai loads no version.

API

• interpret_document(key, *, scope, prompt=None, model=None, client=None) -- returns the interpreted text as a str. scope is required. interpret_document_async is the async form.
• Raises DocumentNotFoundError (a DocumentInterpretError subclass) if the document does not exist, and DocumentInterpretError for any other read/interpretation failure -- including on a persistent 429/5xx after retrying with backoff.
• client lets many calls share one connection; see riffsdk.tasks' API notes above -- both modules default to and can share the same RiffClient/AsyncRiffClient. Every call uses a 130s timeout regardless of the client's own default: this proxies to an LLM call through riff-orchestrator, whose own timeout defaults to 120s.

Email

The AI notice that goes at the end of an email an agent sends. Recipients have to be told when a message was written and sent by an AI agent, and by whom on whose behalf; this builds that notice in one place, so every agent says the same thing and the wording can be corrected without editing each one.

Your tool writes the message and its signature. The notice goes last:

import riffsdk.email.v0 as email

html = body_html + signature_html + email.notice_html(
    "Order Confirmation", on_behalf_of="the Acme purchasing team"
)

which reads:

Sent by Riff.ai Order Confirmation Agent on behalf of the Acme purchasing team. Learn more

Send it on every message, replies included, and use the string as it comes -- rewriting or summarising it defeats the point of having it here.

API

• notice_html(agent, *, on_behalf_of=None, learn_more_url=None) -- one <p> element to concatenate onto an HTML body. It carries its styling inline, because mail clients discard stylesheets.
• notice_text(...) -- the same notice for a message that is not HTML, with the link spelled out. Use it only when the body really is plain text.
• agent is the name without the word "Agent" -- "Order Confirmation" reads as "Riff.ai Order Confirmation Agent". A name that already ends in "Agent" is not repeated.
• on_behalf_of reads straight into the sentence, so pass it as it should appear: "the Acme purchasing team". Left out, the notice names only the agent, which is still a complete disclosure.
• learn_more_url overrides the link, which otherwise points at a page naming this agent. AI_NOTICE_URL is that page without a name.

Versioned like tasks: import the version you wrote against, and it keeps behaving as it did.

Authentication

Set the RIFF_TOKEN environment variable. The SDK picks it up automatically.

API

Clients

• StorageClient -- sync client
• AsyncStorageClient -- async client

Both support: put, get, stat, exists, list, delete, close, and context manager usage.

Models

• ObjectMeta -- metadata for a stored object (key, version, size, content_type, timestamps)
• UploadResult, DownloadResult -- operation results
• ListPage -- paginated listing
• Scope -- access scope (use account_scope(), project_scope(), session_scope())

scope is optional today, both on StorageClient/AsyncStorageClient and on each call that takes one, but will become required in a future version. Leaving it unset anywhere (neither on the client nor on the call) logs a one-time deprecation warning naming the scope it defaulted to (project). Set it explicitly -- as shown throughout this README -- to avoid the warning and be ready for the future version where it's required. The CLI warns the same way when --scope is omitted.

Uploads

• ResumableUpload / AsyncResumableUpload -- multipart resumable uploads for large files
• StorageReader / StorageWriter -- streaming read/write

Content types

content_type is optional on put, upload_file, upload_stream, begin_upload and create_write_stream. When omitted it is derived from the storage key's extension using a table bundled with the SDK, so the result does not depend on the host's /etc/mime.types or the Python version. Unknown extensions fall back to application/octet-stream.

• upload_file prefers the key's extension and falls back to the local filename's -- so upload_file("docs/notes.md", "/tmp/tmpXY123") still stores text/markdown.
• str payloads passed to put() are encoded as UTF-8, and get ; charset=utf-8 appended when the derived type is text/*.
• An explicit content_type= is always used verbatim.

Exceptions

All exceptions inherit from StorageError:

• AuthorisationError
• ObjectNotFoundError
• VersionConflictError
• AlreadyExistsError
• LeaseConflictError
• UploadNotFoundError
• QuotaExceededError
• PartMismatchError
• StorageTransportError

Examples

See the examples/ directory for complete working examples:

• basic_crud.py -- put, get, list, delete
• async_client.py -- async usage with asyncio
• file_upload_download.py -- file uploads with progress
• optimistic_concurrency.py -- version-based conflict handling
• interpret_document.py -- extracting text from a stored PDF/image
• get_task.py -- reading the task board from a trigger tool
• email_notice.py -- appending the AI notice to an email

Development

Requires Python 3.11+ and uv.

uv sync --dev        # Install dependencies
mise run test        # Run tests
mise run lint        # Lint
mise run format      # Format code

See AGENTS.md for full development workflow details.