riven-python

Official Riven Python SDK — an OpenAI-compatible client for the Riven API.

from riven import Riven

client = Riven()  # reads RIVEN_API_KEY from the environment

resp = client.chat.completions.create(
    model="riven-core",
    messages=[{"role": "user", "content": "Say hello from Riven."}],
)
print(resp.choices[0].message.content)
print(resp.citations)  # grounding citations when the model is web-grounded

Install

pip install riven

The importable module is riven.

Quickstart

1. Create a key at console.rivenai.io (starts with rvn_).
2. Export it:

export RIVEN_API_KEY=rvn_...

3. Make a request — this is the SDK equivalent of the curl quickstart:

curl https://api.rivenai.io/v1/chat/completions \
  -H "Authorization: Bearer $RIVEN_API_KEY" \
  -H "Content-Type: application/json" \
  -d '{"model": "riven-core", "messages": [{"role": "user", "content": "Say hello from Riven."}]}'

from riven import Riven

client = Riven()
resp = client.chat.completions.create(
    model="riven-core",
    messages=[{"role": "user", "content": "Say hello from Riven."}],
)
print(resp.choices[0].message.content)

Streaming

stream = client.chat.completions.create(
    model="riven-core",
    messages=[{"role": "user", "content": "Count to five."}],
    stream=True,
)
for chunk in stream:
    delta = chunk.choices[0].delta.content
    if delta:
        print(delta, end="", flush=True)

Grounded responses

Grounded models (e.g. riven-research) return citations and annotations alongside the answer:

resp = client.chat.completions.create(
    model="riven-research",
    messages=[{"role": "user", "content": "What is the population of Tokyo in 2026?"}],
)
print(resp.choices[0].message.content)
for c in resp.citations:
    print(c.title, c.url)

Billing / usage

quota = client.billing.quota()
print(quota.plan_slug, quota.used, quota.limit, quota.reset_at)

# usage helpers attached to every completion
print(resp.usage.prompt_tokens, resp.usage.completion_tokens)
print(resp.grounded, resp.tool_loops)  # from X-Riven-Grounded / X-Riven-Tool-Loops headers

Features

• OpenAI-compatible client pointed at https://api.rivenai.io/v1
• Chat completions + streaming SSE + models listing
• council/query stub (active when the endpoint lands)
• api-key auth (rvn_*), automatic retry on 429 / 5xx with exponential backoff + jitter
• Typed responses via Pydantic models, including grounding fields: citations[], annotations[], grounding_metadata
• Response-header helpers: X-Riven-Tool-Loops, X-Riven-Grounded, X-Riven-Upstream, X-Riven-Quota-*
• Usage + billing / pay-as-you-go status helpers
• Error classes typed to the real gateway error shapes: AuthenticationError, InvalidKeyError, NotFoundError, ValidationError, RateLimitError, QuotaExceededError, APIConnectionError, APIServerError

Configuration

client = Riven(
    api_key="rvn_...",            # default: RIVEN_API_KEY env var
    base_url="https://api.rivenai.io/v1",  # default
    max_retries=3,                 # default
    timeout=60.0,                 # default, seconds
)

Error handling

from riven import Riven, RateLimitError, QuotaExceededError, AuthenticationError

try:
    client.chat.completions.create(model="riven-core", messages=[...])
except RateLimitError as e:
    print("back off", e.retry_after_ms)
except QuotaExceededError as e:
    print("quota resets", e.reset_at, "top up:", e.topup_url)
except AuthenticationError:
    print("check your RIVEN_API_KEY")

Tests

pip install -e ".[dev]"
pytest                        # mock-based unit suite (no key needed)
RIVEN_API_KEY=rvn_... pytest  # adds the live smoke test (skipped without the env var)

License

MIT © 2026 RivenAI