Venice AI Python SDK

This is an unofficial, community-maintained SDK for Venice.ai.
Not affiliated with or endorsed by Venice AI. For official resources visit Venice.ai.

v2.x — fully breaking rewrite over v1.3.x with enterprise-grade features. Review the release notes, the CHANGELOG, and the Migration Guide before upgrading.

Quick Start

Requires Python 3.13+. On earlier versions pip install venice-ai resolves to v1.3.x
without an error. Pin the major version so you get a clear failure instead of the wrong SDK.

pip install 'venice-ai>=2'
export VENICE_API_KEY="your-api-key-here"

import asyncio
from venice_ai import VeniceClient, UserMessage

async def main():
    async with VeniceClient() as client:  # reads VENICE_API_KEY from env
        model = await client.models.resolve_chat()
        response = await client.chat.completions.create(
            model=model,
            messages=[UserMessage(content="Hello!")],
        )
        print(response.choices[0].message.content)

asyncio.run(main())

Migrating from v1.x? See the Migration Guide.

Command Line Interface

pip install 'venice-ai[cli]>=2'

venice chat start                  # Interactive chat
venice image generate "..."        # Image generation
venice image multi-edit -p "..."   # Multi-image edit
venice models                      # Browse models
venice characters reviews <slug>   # Character reviews
venice account keys rate-limits    # Per-model RPM/TPM limits
venice configure                   # Setup wizard

Features: streaming chat with 6 animation modes, image generation with 11+ parameters, model discovery, rich terminal UI. See the CLI Reference for full CLI documentation.

Build with Claude Code

Four Claude Code skills ship with the SDK. Install them into the current project with venice skills install (or venice skills install --global for ~/.claude/skills/); list them with venice skills list. They auto-load when their trigger contexts match — venice chat, venice image, venice rate limit, venice x402 — and steer Claude toward idiomatic v2 code (dynamic model resolution, async with stream:, run_with_tools, client.gather(max_concurrency=N), top_up_with, etc.) instead of OpenAI-style or v1 patterns.

venice skills install            # → ./.claude/skills/
venice skills install --global   # → ~/.claude/skills/
venice skills list               # show bundled skills + install state

Catalog: venice-ai (chat / streaming / tools / structured output), venice-ai-multimodal (image / audio / video / music), venice-ai-production (retries / rate limits / cost tracking / observability), venice-ai-x402 (wallet auth / SIWE / on-chain top-up). See tools/skills/README.md for the full catalog and validation tooling.

Core Features

Chat Completions

from venice_ai import UserMessage, SystemMessage

async with VeniceClient() as client:
    model = await client.models.resolve_chat()
    response = await client.chat.completions.create(
        model=model,
        messages=[
            SystemMessage(content="You are a helpful assistant."),
            UserMessage(content="Explain quantum computing simply"),
        ],
        temperature=0.7,
        max_completion_tokens=500,
    )
    print(response.choices[0].message.content)

-> examples/chat/simple_chat.py

Streaming

model = await client.models.resolve_chat()
async with await client.chat.completions.stream(
    model=model,
    messages=[UserMessage(content="Tell me a story")],
    max_completion_tokens=200,
) as stream:
    async for text in stream.text_deltas():
        print(text, end="", flush=True)

For the assembled response: response = await stream.collect().

-> examples/chat/streaming_chat.py

Synchronous Usage

from venice_ai import SyncVeniceClient, UserMessage

with SyncVeniceClient() as client:
    model = client.models.resolve_chat()
    response = client.chat.completions.create(
        model=model,
        messages=[UserMessage(content="Hello!")],
    )
    print(response.choices[0].message.content)

Streams returned by SyncVeniceClient iterate synchronously (for chunk in stream:).

Function Calling

from venice_ai import tool_from_function

def get_weather(location: str) -> str:
    """Get current weather for a location."""
    ...

model = await client.models.resolve_chat(require_function_calling=True)
response = await client.chat.completions.create(
    model=model,
    messages=[UserMessage(content="What's the weather in NYC?")],
    tools=[tool_from_function(get_weather)],
    tool_choice="auto",
)

tool_from_model(MyPydanticModel) is also available for richer schemas.

-> examples/chat/tool_calling.py

Image Generation

image_model = await client.models.resolve_image()
response = await client.image.create(
    model=image_model,
    prompt="A serene mountain landscape at sunset",
    width=512, height=512,
    enable_web_search=True,  # optional; supported models pull in recent web context
)
response.save("generated.png")          # single image
# response.save_all("output_dir")        # all images

Pass-through fields on image.multi_edit() now include model=..., which the
SDK forwards to the API as modelId (previously dropped silently).

-> examples/image/text_to_image.py | -> examples/image/web_search.py

Text-to-Speech

from venice_ai.types.enums import Voice, ResponseFormat

tts_model = await client.models.resolve_tts()
response = await client.audio.create_speech(
    model=tts_model,
    input="Hello! Welcome to Venice AI.",
    voice=Voice.AF_ALLOY,
    response_format=ResponseFormat.MP3,
)
response.save("speech.mp3")

-> examples/audio/text_to_speech.py

Embeddings

embedding_model = await client.models.resolve_embedding()
response = await client.embeddings.create(
    model=embedding_model,
    input=["Text 1", "Text 2", "Text 3"],
)
embedding = response.data[0].embedding

-> examples/embeddings/basic_embeddings.py

Video Generation

video_model = await client.models.resolve_video()
job = await client.video.run(
    model=video_model,
    prompt="A drone shot of the Venice canals at sunrise",
    duration_seconds=5,
    aspect_ratio="16:9",
    resolution="1080p",
)
async with job:
    status = await job.wait()
    await job.download("canals.mp4", status)

Advanced body fields on submit() (all optional):
upscale_factor, end_image_url, audio_url, video_url,
reference_image_urls (up to 9), elements (up to 4 Kling-O3 structured
characters), scene_image_urls (up to 4). For the dedicated
topaz-video-upscale model, pass video_url + upscale_factor (1/2/4)
instead of resolution. quote() accepts only the pricing-relevant
subset (model, duration_seconds, aspect_ratio, resolution, upscale_factor,
audio, video_url) per the API spec — prompt text and reference images
don't affect price.

-> examples/video/text_to_video.py | -> examples/video/advanced_fields.py | -> examples/video/upscale.py

Model Selection

chat_model = await client.models.resolve_chat(
    preferred_models=["llama-3.3-70b", "qwen-2.5-72b"],
    require_function_calling=True,
)
image_model = await client.models.resolve_image()
embedding_model = await client.models.resolve_embedding()

# Capability filters via the unified entry point:
vision_model = await client.models.resolve(type="chat", require_vision=True)

Venice-Specific Features

Character Personalities

from venice_ai import VeniceParameters

model = await client.models.resolve_chat()
response = await client.chat.completions.create(
    model=model,
    messages=[UserMessage(content="What is wisdom?")],
    venice_parameters=VeniceParameters(character_slug="socrates"),
)

Web Search

model = await client.models.resolve_chat()
response = await client.chat.completions.create(
    model=model,
    messages=[UserMessage(content="Latest AI news?")],
    venice_parameters=VeniceParameters(enable_web_search="on", enable_web_citations=True),
)

Web Scrape, Search & Text Parsing (Augment)

# Scrape a URL and get markdown back
page = await client.augment.scrape(url="https://example.com")
print(page.content)

# Structured web search (Brave default; Google also supported)
hits = await client.augment.search(query="latest AI news", limit=5)
for r in hits.results:
    print(r.title, r.url)

# Parse a document (PDF / DOCX / XLSX / TXT, ≤ 25 MB)
parsed = await client.augment.parse_text(file="report.pdf")
print(parsed.text, parsed.tokens)

-> examples/augment/scrape.py | -> examples/augment/search.py | -> examples/augment/text_parser.py

x402 Wallet Billing (optional)

The x402 billing endpoints use Ethereum wallet auth (SIWE / EIP-4361 on
Base) instead of Bearer tokens. Install the optional extra to pick up
eth-account + siwe:

pip install 'venice-ai[x402]'

from venice_ai.auth.x402 import X402Auth

auth = X402Auth(private_key=os.environ["X402_WALLET_PRIVATE_KEY"])

async with VeniceClient() as client:
    balance = await client.x402.balance(auth=auth)
    print(f"${balance.data.balanceUsd} on {auth.wallet_address}")

    txns = await client.x402.transactions(auth=auth)
    for t in txns.data.transactions[:5]:
        print(t.createdAt, t.type, t.amount)

    # Empty POST discovers x402 payment requirements; the response surfaces
    # as a 402 APIError whose body carries the accept spec.
    await client.x402.top_up()  # or top_up(payment_header=<signed b64>)

Prefer one-call top-ups? client.x402.top_up_with(auth=auth, amount_usdc=5.0) runs the
full EVM probe → sign → submit flow for you. To settle from a Solana wallet instead,
install venice-ai[x402-solana] and use SolanaX402Auth with top_up_with_solana:

from venice_ai.auth.x402_solana import SolanaX402Auth

auth = SolanaX402Auth(private_key=os.environ["X402_SOLANA_SECRET"])  # base58 secret
async with VeniceClient() as client:
    await client.x402.top_up_with_solana(auth=auth, amount_usdc=5.0)

-> examples/x402/balance.py | -> examples/x402/transactions.py | -> examples/x402/top_up.py

Confidential Compute (TEE / E2EE, optional)

Venice's e2ee-* models run in a Trusted Execution Environment with
client-side end-to-end encryption: attest the enclave, then encrypt each
message under a key only the enclave can derive (secp256k1 ECDH -> HKDF-SHA256
-> AES-256-GCM). Install the optional extra (cryptography):

pip install 'venice-ai[e2ee]'

# One-shot: just turn on E2EE for an e2ee-* model. The SDK attests the enclave,
# opens a session, and encrypts/decrypts transparently.
model = await client.models.resolve_chat()  # pick an e2ee-* model
response = await client.chat.completions.create(
    model=model,
    messages=[UserMessage(content="Confidential question")],
    e2ee=True,  # equivalent: venice_parameters=VeniceParameters(enable_e2ee=True)
)

# Or drive the lifecycle yourself for low-level control:
attestation = await client.tee.get_attestation(model=model)  # fail-closed verify
with await client.tee.open_session(model=model) as session:
    headers = session.request_headers()
    blob = session.encrypt_message("Hello, confidential world.")
    # ... POST the encrypted content with `headers`; decrypt streamed deltas via
    #     session.decrypt_chunk(delta_hex)

Full client-side TDX verification ([e2ee-verify])

The default path is baseline: it trusts Venice's server-side verified
claim and does not independently verify the Intel TDX quote. For threat models
that include a malicious Venice operator, install the [e2ee-verify] extra and
pass a DcapTdxVerifier, which verifies the raw quote's ECDSA signature and PCK
certificate chain to a pinned Intel SGX Root CA, the TCB status, the non-debug
flag, the REPORTDATA key binding, the RTMR event-log replay, and the dstack
compose-hash — all offline:

pip install 'venice-ai[e2ee-verify]'   # dcap-qvl (+ cryptography)

from venice_ai.tee import DcapTdxVerifier, TeeOptions

model = await client.models.resolve_chat()  # pick an e2ee-* model

# Fetch Intel-signed collateral once (the only network touch); verify() is offline.
verifier = await DcapTdxVerifier.with_fetched_collateral(
    probe_quote=(await client.tee.get_attestation(model=model)).intel_quote,
)

# Run the full verifier as part of attestation / session open:
session = await client.tee.open_session(model=model, verifier=verifier)

# Or engage it through chat E2EE:
response = await client.chat.completions.create(
    model=model,
    messages=[UserMessage(content="Confidential question")],
    e2ee=TeeOptions(verifier=verifier),
)

What it proves (Tier B). By default DcapTdxVerifier proves the model
runs on a genuine, non-debug Intel TDX enclave running a self-consistent
dstack workload. It does not by itself prove this is the legitimate
Venice workload — there are no published reference measurements today. Supply
expected_measurements / expected_compose_hash from an independent source to
pin workload identity (Tier A). TCB status is fail-closed reject-by-default
(tcb_policy="advisory" to accept hardening-needed statuses with advisories).
NVIDIA GPU attestation is not yet shipped.

Reasoning Controls & Cost Tracking

# Reasoning effort tier — top-level parameter on chat.completions.create().
# Seven tiers (per-model support): none / minimal / low / medium / high / xhigh / max.
response = await client.chat.completions.create(
    model=await client.models.resolve_chat(require_reasoning=True),
    messages=[UserMessage(content="Prove √2 is irrational.")],
    reasoning_effort="max",
)

# Nested form with summary verbosity. Top-level reasoning_effort takes
# precedence over reasoning.effort when both are set.
from venice_ai import ReasoningConfig
response = await client.chat.completions.create(
    model=reasoning_model,
    messages=[UserMessage(content="Explain quantum entanglement.")],
    reasoning=ReasoningConfig(effort="high", summary="concise"),
)

# Show/hide raw thinking blocks via venice_parameters
venice_params = VeniceParameters(strip_thinking_response=False, disable_thinking=False)

# Cost tracking
from venice_ai import calculate_completion_cost
cost = calculate_completion_cost(response, model_pricing=None)
print(f"Cost: ${cost['usd']:.4f}")

Configuration

Installation Options

Quote the specifier — [...] is a glob in zsh, and the >=2 floor turns an
incompatible Python into a clear pip error instead of a silent v1.3.x install.

pip install 'venice-ai>=2'                # Core
pip install 'venice-ai[cli]>=2'           # CLI tools
pip install 'venice-ai[redis]>=2'         # Redis backend
pip install 'venice-ai[enterprise]>=2'    # Enterprise (redis + prometheus + otel)
pip install 'venice-ai[adaptive]>=2'      # Adaptive rate limiting
pip install 'venice-ai[x402]>=2'          # x402 wallet auth (eth-account + siwe)
pip install 'venice-ai[x402-solana]>=2'   # x402 Solana USDC top-up (solders)
pip install 'venice-ai[e2ee]>=2'          # TEE client-side E2EE (cryptography)
pip install 'venice-ai[e2ee-verify]>=2'   # Full client-side TDX quote verification (dcap-qvl)
pip install 'venice-ai[all]>=2'           # Everything

Client Setup

# Minimal (reads VENICE_API_KEY from environment)
async with VeniceClient() as client: ...

# Explicit
async with VeniceClient(api_key="your-key") as client: ...

# Factory with full configuration
from venice_ai import VeniceClientFactory, VeniceAIConfig
from venice_ai.core.config import BackendConfig, BackendType, HttpClientConfig, SchedulerConfig, SchedulerMode

config = VeniceAIConfig(
    backend=BackendConfig(backend_type=BackendType.MEMORY),
    http_client=HttpClientConfig(timeout=60.0, max_connections=50),
    scheduler=SchedulerConfig(mode=SchedulerMode.BASIC)
)
client = VeniceClientFactory.create_client(config=config, api_key=os.getenv("VENICE_API_KEY"))

Environment Variables

Configure with VENICE_ prefix (double underscores for nesting):

export VENICE_SCHEDULER__MODE=intelligent
export VENICE_BACKEND__REDIS__REDIS_URL=redis://localhost:6379
export VENICE_HTTP_CLIENT__TIMEOUT=60.0

Note: Env var auto-loading requires pydantic-settings: pip install 'venice-ai[enterprise]>=2'

Advanced Features

Rate limiting, distributed state, monitoring, observability, and performance tuning are covered in Advanced Features.

API Resources

Type Safety

The SDK uses Pydantic v2 models throughout:

from venice_ai.types.api import UserMessage, SystemMessage, AssistantMessage, ToolMessage
from venice_ai.types.api.requests import VeniceParameters, StreamOptions
from venice_ai.types.api.requests.common import Tool, ToolFunction
from venice_ai.types import JSONSchemaFormat
from venice_ai.types.chat import ChatCompletionChunk
from venice_ai.types.audio import Voice, ResponseFormat

Testing

from venice_ai import create_test_venice_client
from venice_ai.core.config import SchedulerMode

async with create_test_venice_client(api_key="test-key", scheduler_mode=SchedulerMode.BASIC) as client:
    response = await client.chat.completions.create(...)

make test           # All tests (parallel)
make test-unit      # Unit tests only
make test-e2e       # E2E tests (requires API key)
make test-verbose   # With coverage

Best Practices

1. Always use context managers for proper cleanup
2. Handle errors with specific exception types (RateLimitError, AuthenticationError, etc.)
3. Monitor rate limits via response.response_rate_limits
4. Use environment variables for API keys (never hardcode)
5. Use Redis backend for production multi-instance deployments
6. Use streaming for long responses to reduce time to first token

Requirements

• Python 3.13+
• Core deps: aiohttp (>=3.13.4,<3.15), pydantic (^2.13.4)
• Platform: Linux, macOS, Windows

See Installation Options for optional dependencies.

Contributing

git clone https://github.com/sethbang/venice-ai.git && cd venice-ai
poetry install
make test

Follow PEP 8, use type hints, write tests, and submit a PR.

Support

• Documentation
• Issue Tracker
• Examples
• Changelog

License

MIT License — see LICENSE.