Python SDK

veqen for Python. Wrap OpenAI, Anthropic, Google and any OpenAI-compatible client, for text, embeddings, images and audio; tag calls, set budgets, record any call by hand and price models offline.

pip install veqen

Python 3.10+, no dependencies. Source: github.com/Keyveve/veqen-sdks, MIT licensed.

Configure

import veqen

veqen.init()  # optional: the first wrapped call configures from the environment

Wrap a client

from anthropic import Anthropic
from google import genai
from openai import AsyncOpenAI, OpenAI

openai_client = veqen.wrap(OpenAI(), feature="search")
async_client = veqen.wrap(AsyncOpenAI())
claude = veqen.wrap(Anthropic(), residency="us")      # US-only processing is priced at 1.1x
gemini = veqen.wrap(genai.Client())

Keyword arguments to wrap() become default tags for every call through that client. The wrapper instruments:

Image calls are priced per image at the requested size and quality (or by image tokens, when reported), speech by characters, and transcription by seconds of audio or tokens.

An OpenAI client pointed at another provider is priced as that provider, read from its base URL: Together, Fireworks, Groq, DeepInfra, Cerebras, Azure AI Foundry and the rest of the price book's hosts. For your own server, name the provider:

together = veqen.wrap(OpenAI(base_url="https://api.together.xyz/v1", api_key=...), project="research")
local = veqen.wrap(OpenAI(base_url="http://vllm.internal:8000/v1"), provider="self-hosted")

Everything else on the client passes straight through. Streams are priced when they finish; for OpenAI chat streams, the wrapper turns on stream_options.include_usage so the provider reports usage. Calls whose response carries no usage aren't recorded.

Tag calls

with veqen.tags(customer=org.id, user=user.id):
    answer = openai_client.responses.create(model="gpt-5-mini", input=question)

Tags nest, and they follow asyncio tasks and asyncio.to_thread calls started inside the block (they live in a ContextVar). Threads you start yourself don't inherit them: set tags inside the thread. See Tags.

Budgets

veqen.budget(5.00, customer="*")              # every customer: $5 a month, then BudgetExceeded
veqen.budget(200, feature="agent", period="day", mode="warn")
veqen.budget(1_000, project="eval-harness")       # scopes: customer, feature, user, project

try:
    openai_client.responses.create(...)
except veqen.BudgetExceeded as exc:
    print(exc.scope, exc.value, exc.limit_usd)

Local budgets count spend in this process. Budgets created in the dashboard are shared by every process using your key; the SDK refreshes them every 30 seconds and enforces them the same way. See Budgets.

Record a call by hand

For calls the wrapper can't see (batch results, gateways, other languages):

veqen.record(
    "claude-sonnet-4-5",
    provider="anthropic",
    input_tokens=1200, output_tokens=300, cache_read_tokens=8000,
    batch=True,
    customer="acme", feature="nightly-summary",
)

Models billed per unit take units (images, seconds of audio or video, characters, queries or pages) and, for images and video, the variant:

veqen.record("veo-3.1-generate-001", provider="google", units=8, variant="1080p", project="ads")
veqen.record("flux-pro-1.1", provider="bfl", units=4, feature="catalog")
veqen.record("nova-3", provider="deepgram", units=1800, project="call-analytics")   # 30 minutes

Pass cost_usd= for a model the price book doesn't know, or for a call that reports a cost but no usage (fal and Replicate jobs often do); it's kept as reported. Prices for your own models go in custom prices.

Price without recording

veqen.cost("gpt-4o", 1000, 500)                        # 0.0075
veqen.cost("gpt-image-1", units=1, variant="high/1024x1024")
veqen.cost("whisper-1", units=600)                       # ten minutes of audio
veqen.cost("claude-sonnet-4-5", 2000, 400, cache_read_tokens=10_000, residency="us")
price = veqen.lookup("us.anthropic.claude-sonnet-4-5-20250929-v1:0")   # Bedrock ids resolve too
price.rates["standard"], price.retires

Serverless and shutdown

Events are sent every two seconds and at interpreter exit. In AWS Lambda, Cloud Functions or similar, call veqen.flush() before returning: it waits up to five seconds and returns True once everything is sent.

Integrations

LangChain, LiteLLM and the OpenAI Agents SDK: see Framework integrations.