veniceresch 0.6.1

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).

What "typed" means here

You get typed resource methods (autocompleted kwargs, typed return wrappers) while the client stays tolerant of Venice API drift. The response wrappers are deliberately shallow: top-level fields are typed, but nested collections are exposed as dict[str, Any] rather than strict generated models — which is why the examples below index into dicts (response.choices[0]["message"]["content"], created.data["apiKey"]). That keeps a renamed or dropped upstream field from breaking your call mid-stream. Unknown fields are preserved on .model_extra instead of raising. When you do want the strict nested form, validate an element explicitly — e.g. ModelResponse.model_validate(result.data[0]). This is an alpha-stage choice; don't expect deep end-to-end type safety on nested payloads.

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 is its own Venice API (not an OpenAI-compatibility alias for chat), but it streams through the same await-then-iterate 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.