veniceresch 0.6.0

Typed, async-first Python client for the Venice.ai API.

veniceresch

A typed, async-first Python client for the Venice.ai API.

Why this exists

• Venice has no official Python SDK. github.com/veniceai publishes a CLI, docs, MCP server, and x402 client — no Python library.
• But they do publish an OpenAPI 3.0 spec at github.com/veniceai/api-docs/blob/main/swagger.yaml.
• This project treats that spec as the source of truth for types and wraps it in a small hand-written httpx client with Venice-specific ergonomics (venice_parameters, typed errors, binary response handling, video polling helper).

Install

pip install veniceresch

Quickstart

import asyncio
from veniceresch import AsyncVeniceClient

async def main():
    async with AsyncVeniceClient(api_key="...") as client:   # or set VENICE_API_KEY
        response = await client.chat.create(
            model="llama-3.3-70b",
            messages=[{"role": "user", "content": "Hello!"}],
            venice_parameters={"include_venice_system_prompt": False},
        )
        print(response.choices[0]["message"]["content"])

asyncio.run(main())

The OpenAI-namespaced form works too — same call, identical HTTP:

await client.chat.completions.create(
    model="llama-3.3-70b",
    messages=[{"role": "user", "content": "Hello!"}],
)

The most common OpenAI-compatible fields (temperature, top_p, n, stop, max_tokens, frequency_penalty, presence_penalty, seed, tools, tool_choice, response_format, logprobs) are named parameters so your IDE can autocomplete them. Everything else Venice accepts (min_p, repetition_penalty, reasoning_effort, prompt_cache_key, top_k, …) still flows through as **extra and is forwarded verbatim:

await client.chat.create(
    model="llama-3.3-70b",
    messages=[{"role": "user", "content": "Summarize today's news."}],
    temperature=0.5,
    max_tokens=512,
    response_format={"type": "json_object"},
    tools=[{"type": "function", "function": {"name": "get_time"}}],
    tool_choice="auto",
    # Venice-specific extras still accepted via **extra:
    min_p=0.05,
    reasoning_effort="medium",
)

Streaming

Pass stream=True to get an async iterator. await the call, then async for the iterator (same contract as the openai SDK):

stream = await client.chat.completions.create(
    model="llama-3.3-70b",
    messages=[{"role": "user", "content": "Tell me a story."}],
    stream=True,
)
async for event in stream:
    delta = event.choices[0]["delta"].get("content", "")
    print(delta, end="", flush=True)

client.chat.stream(...) is an explicit alias; it uses the same await-then-iterate contract.

/responses streams through the same contract:

stream = await client.responses.create(
    model="venice-reasoning-1",
    input="Tell me a short joke.",
    stream=True,
)
async for event in stream:
    # Venice's per-chunk schema isn't in their swagger yet; known
    # identifier fields (id/object/created_at/model) are typed, and
    # everything else (deltas, block events, sequence numbers, …)
    # lands on event.model_extra.
    print(event.model_extra)

client.responses.stream(...) is the explicit alias.

Image: generate, edit, multi-edit

# JSON response (base64-encoded images in the dict)
result = await client.image.generate(model="flux-dev", prompt="a red cube", width=1024, height=1024)

# Raw bytes (PNG/JPEG/WebP)
png_bytes = await client.image.generate_binary(model="flux-dev", prompt="a red cube")

# Edit — accepts bytes, Path, base64 string, or URL
edited = await client.image.edit(image=Path("cat.png"), prompt="make it blue", model="flux-edit")

# Multi-edit (up to 3 images)
combined = await client.image.multi_edit(
    images=[img1_bytes, img2_bytes],
    prompt="merge them",
    model_id="flux-multi-edit",
)

# Upscale / background-remove
upscaled = await client.image.upscale(image=source_png, scale=2.0)
cutout = await client.image.background_remove(image_url="https://x/pic.png")

OpenAI-compatible image alias

Drop-in replacement for openai.images.generate(...). Hits Venice's separate /images/generations endpoint, which accepts the full OpenAI parameter set (many are accepted for compatibility but not used by Venice — see the endpoint's swagger for the list). n is clamped to 1.

result = await client.images.generate(
    prompt="a red cube",
    n=1,
    size="1024x1024",
    response_format="b64_json",
)
print(result.data[0]["b64_json"])

client.image (singular) is the primary Venice-native image surface and covers more endpoints (edit, multi-edit, upscale, background-remove, styles). client.images (plural) exists only for OpenAI compatibility.

Video: queue + poll

queued = await client.video.queue(model="video-v1", prompt="a cat", duration="5s")

# Poll until done (raises VeniceVideoTimeoutError at timeout):
result = await client.video.wait_for_completion(
    model="video-v1", queue_id=queued.queue_id, timeout_s=600, poll_interval_s=2.0,
)
print(result.status)  # "COMPLETED"

# Fetch the MP4 bytes:
mp4 = await client.video.retrieve_binary(model="video-v1", queue_id=queued.queue_id)

wait_for_completion waits for a terminal state, not necessarily a successful one: by default it returns the final response for any non-PROCESSING status (so unknown terminal statuses are tolerated). Pass raise_on_failed=True to instead raise VeniceVideoFailedError / VeniceAudioFailedError on a known failure status (FAILED, CANCELLED, ERROR); the final response is on the error's .result.

Queue / retrieve / quote / complete / transcribe all return typed Pydantic models (VideoQueueResponse, VideoRetrieveResponse, etc.). Attribute access everywhere — queued.queue_id and result.status, not dict indexing.

Audio: TTS, transcription, queued generation

mp3_bytes = await client.audio.create_speech(input="Hello world", voice="nova")
transcript = await client.audio.transcribe(file=Path("clip.wav"), model="whisper-1")
print(transcript.text)

Every file= argument (here, voice cloning, and augment.parse/parse_text) accepts raw bytes, a path (str or Path), or a binary file-like object — e.g. transcribe(file=open("clip.wav", "rb"), ...) or a BytesIO. Paths and file handles are streamed to the server rather than buffered in memory; a handle you pass in is left open for you to close. (The base64 image.* endpoints accept the same input forms but must buffer in full — pass an https:// URL to skip uploading large local files.)

Queued audio generation mirrors video — client.audio.queue(...) → AudioQueueResponse, poll client.audio.retrieve(...) / wait_for_completion(...) → AudioRetrieveResponse, then retrieve_binary(...) for the audio bytes.

Voice cloning. client.audio.create_cloned_voice(...) uploads a sample (multipart) and returns a vv_<id> handle to reuse as the voice argument of create_speech with the same model. Handles expire after ~7 days.

voice = await client.audio.create_cloned_voice(
    file=Path("sample.mp3"), model="tts-chatterbox-hd",
)
mp3 = await client.audio.create_speech(
    input="Now in the cloned voice.", voice=voice.id, model="tts-chatterbox-hd",
)

Models / embeddings / billing

models = await client.models.list(type="video")
emb = await client.embeddings.create(input="hello", model="embed-v1")
balance = await client.billing.balance()

Augment: scrape, search, parse

scraped = await client.augment.scrape(url="https://example.com")
results = await client.augment.search(query="venice ai", limit=5)

# Parse a PDF/DOCX/XLSX/text file — JSON form returns {text, tokens}
parsed = await client.augment.parse(file=Path("doc.pdf"))
print(parsed.text, parsed.tokens)

# Plain-text form returns a str directly (Accept: text/plain)
raw_text = await client.augment.parse_text(file=Path("doc.pdf"))

Characters

listing = await client.characters.list(sort_by="featured", limit=20)
for item in listing.data:
    print(item["slug"], item["name"])

character = await client.characters.get("lucy")
reviews = await client.characters.reviews("lucy", page=1, page_size=20)
print(reviews.summary, reviews.pagination)

Query parameters use Python snake_case (is_adult, is_pro, is_web_enabled, sort_by, sort_order, model_id, page_size); the client translates them to the camelCase Venice expects.

API key management

client.api_keys.* wraps Venice's /api_keys/* routes — these normally require an admin-scope key (inference keys don't grant access to key management, so expect VeniceAuthError on the wrong key type).

keys = await client.api_keys.list()
detail = await client.api_keys.get("k1")
created = await client.api_keys.create(
    api_key_type="INFERENCE",
    description="my bot",
    consumption_limit={"usd": 50, "diem": 10},
)
print(created.data["apiKey"])  # only returned once — save it now

await client.api_keys.update(id="k1", description="renamed")
await client.api_keys.delete("k1")

limits = await client.api_keys.rate_limits()
exceedances = await client.api_keys.rate_limits_log()

The /api_keys/generate_web3_key endpoints mint a key from a wallet signature instead of an admin token; they skip the default Authorization header automatically. Bring your own signer:

challenge = await client.api_keys.generate_web3_key_challenge()
signature = my_wallet.sign(challenge.data["token"])  # your signing code

minted = await client.api_keys.generate_web3_key(
    api_key_type="INFERENCE",
    address="0x...",
    signature=signature,
    token=challenge.data["token"],
)
print(minted.data["apiKey"])

x402 (wallet-paid credit)

client.x402.* wraps Venice's /x402/* routes. These use wallet-based auth — not the bearer token — so this SDK accepts the signed header payloads as strings and sends them verbatim. We don't bundle a wallet signer; generate the SIWE / X-402-Payment payloads with whatever tooling you already use.

balance = await client.x402.balance("0xabc...", siwx_header=siwe_payload)
print(balance.data["balanceUsd"], balance.data["canConsume"])

history = await client.x402.transactions(
    "0xabc...", siwx_header=siwe_payload, limit=25,
)

top_up returns the credited-balance payload on success. Calling it with no header triggers Venice's 402 discovery flow, which raises VeniceX402PaymentRequiredError carrying the accepted payment options on .accepts:

from veniceresch import VeniceX402PaymentRequiredError

try:
    await client.x402.top_up()  # no header → discovery
except VeniceX402PaymentRequiredError as exc:
    for option in exc.accepts:
        print(option["network"], option["asset"], option["amount"], option["payTo"])
    signed = my_wallet.sign_x402_payment(exc.accepts[0])
    credited = await client.x402.top_up(payment_header=signed)
    print(credited.data["newBalance"])

Crypto JSON-RPC proxy

client.crypto proxies JSON-RPC 2.0 calls to supported chains, billed per credit. networks() lists valid network slugs (public, no auth). rpc() mirrors the request shape — a dict for a single call, a list for a batch of up to 100. Pass siwx_header= to pay with an x402 wallet instead of the default API key, or idempotency_key= for safe retries.

slugs = (await client.crypto.networks()).networks

chain_id = await client.crypto.rpc(
    "ethereum-mainnet",
    {"jsonrpc": "2.0", "method": "eth_chainId", "params": [], "id": 1},
)
print(chain_id["result"])

Per-request JSON-RPC failures come back as HTTP 200 with an error field on the response item — they do not raise.

Auto-pagination

Four list endpoints ship companion iter_* methods that walk every page for you. They return an AsyncPaginator (or Paginator on the sync client) — iterate it to get one item at a time, or call .iter_pages() to get whole response objects:

async for tx in client.x402.iter_transactions("0xabc...", siwx_header=siwe_payload):
    print(tx["id"], tx["amount"])

# Or page-by-page:
async for page in client.x402.iter_transactions(
    "0xabc...", siwx_header=siwe_payload
).iter_pages():
    print(f"{len(page.data['transactions'])} transactions on this page")

The same pattern works for client.characters.iter_list(...), client.characters.iter_reviews(slug), and client.billing.iter_usage(...). No HTTP request fires until iteration starts. The single-page methods (transactions, list, reviews, usage) still work unchanged.

Error handling

Every failure raises a subclass of VeniceError. HTTP responses map to VeniceAPIError; transport-level failures (DNS, TLS, timeouts) map to VeniceConnectionError / VeniceTimeoutError:

Exception	When
VeniceAuthError	401 — bad or missing API key
VeniceInsufficientBalanceError	402 — balance exhausted
VeniceX402PaymentRequiredError	402 from an x402 endpoint — body is an x402 discovery payload (x402_version, accepts), not an error
VeniceValidationError	400 / 422 — bad request shape
VeniceProviderContentPolicyError	422 — upstream provider rejected on content policy (recommended_model, credits_refunded); detected by body shape
VeniceNotFoundError	404
VenicePayloadTooLargeError	413 — request payload exceeds Venice's size limit
VeniceRateLimitError	429
VeniceServerError	5xx
VeniceContentViolationError	body contained suggested_prompt (any status)
VeniceConnectionError	DNS / TLS / connection reset / proxy failure
VeniceTimeoutError	request or response timed out

from veniceresch import (
    VeniceConnectionError,
    VeniceContentViolationError,
    VeniceRateLimitError,
    VeniceServerError,
    VeniceTimeoutError,
)

# Retriable failures — no httpx imports needed:
try:
    await client.chat.create(model="m", messages=[...])
except (VeniceRateLimitError, VeniceServerError,
        VeniceConnectionError, VeniceTimeoutError):
    ...  # back off and retry

try:
    await client.image.generate(model="m", prompt="...")
except VeniceContentViolationError as exc:
    if exc.suggested_prompt:
        retry = await client.image.generate(model="m", prompt=exc.suggested_prompt)

VeniceConnectionError.__cause__ is the underlying httpx exception if you need to introspect it.

Sync client

from veniceresch import VeniceClient

with VeniceClient(api_key="...") as client:
    for event in client.chat.completions.create(
        model="...", messages=[...], stream=True,
    ):
        ...

Sync stream() returns an iterator directly (no await — that form is async-only).

Development

Types come from Venice's upstream OpenAPI spec, regenerated via bash scripts/regen_types.sh (pulls the latest swagger, runs datamodel-code-generator, writes src/veniceresch/_generated.py). Pass --offline to use the pinned vendor/venice-swagger.yaml instead.

pip install -e ".[dev]"
ruff check . && ruff format --check .
mypy src/veniceresch
pytest                                    # unit tests (offline, respx-mocked)
VENICE_API_KEY=... pytest tests/integration -m integration  # smoke

Endpoint coverage

Group	Covered	Gap
chat	/chat/completions	—
responses	/responses (streaming + non-streaming)	—
image	/image/generate, /image/edit, /image/multi-edit, /image/upscale, /image/background-remove, /image/styles, /images/generations (OpenAI alias via client.images.generate)	—
video	/video/queue, /video/retrieve, /video/quote, /video/complete, /video/transcriptions	—
audio	/audio/speech, /audio/voices (voice cloning), /audio/transcriptions, /audio/queue, /audio/retrieve, /audio/quote, /audio/complete	—
models	/models, /models/traits, /models/compatibility_mapping	—
embeddings	/embeddings	—
billing	/billing/balance, /billing/usage, /billing/usage-analytics	—
augment	/augment/scrape, /augment/search, /augment/text-parser	—
characters	/characters, /characters/{slug}, /characters/{slug}/reviews	—
api_keys	/api_keys (list/create/update/delete), /api_keys/{id}, /api_keys/rate_limits, /api_keys/rate_limits/log, /api_keys/generate_web3_key (GET + POST)	—
x402	/x402/balance/{walletAddress}, /x402/top-up, /x402/transactions/{walletAddress}	—
crypto	/crypto/rpc/networks, /crypto/rpc/{network} (JSON-RPC proxy)	—

All 44 paths in Venice's current OpenAPI spec are covered. The x402 and web3 endpoints use wallet-based auth — this SDK accepts the signed header payloads you produce (SIWE / X-402-Payment) and forwards them verbatim; it does not bundle a wallet signer.

Non-goals (unchanged from v0.1): retry/backoff, CLI, SSE parsing beyond decoded JSON events, tool-use schema builders.

License

MIT.