usageflow-vibe (beta)

usageflow-vibe is one Python client for Anthropic and OpenAI that checks every call against your UsageFlow limits and policies before it runs, then records the real token usage.

It is separate from the Flask/FastAPI route middleware: use it when you call LLMs from your own code and want to meter and govern those calls per customer.

Beta: this package is new and its APIs may still change between minor versions. If you hit an issue, please report it.

Usage is recorded asynchronously right after each call, so a balance you read back from credits() can lag by a moment; withdraw, credit, and close return as soon as the request is sent, not once it has been applied.

Install

pip install usageflow-vibe
export USAGEFLOW_API_KEY="your-api-key"
export OPENAI_API_KEY="..."      # only the provider(s) you call
export ANTHROPIC_API_KEY="..."

Python 3.9+. The only dependency is websocket-client; provider calls use the standard library.

Chat

from usageflow.vibe import VibeClient, Message, RejectionError

client = VibeClient()  # reads USAGEFLOW_API_KEY

try:
    result = client.chat(
        identity="cust_acme",          # the customer you meter
        workflow="support-agent",      # optional: the slug of a Vibe policy
        model="claude-sonnet-5",       # provider inferred: claude-* → Anthropic, else OpenAI
        messages=[Message("user", "Summarize this ticket.")],
        customer_metadata={"plan": "pro"},  # optional: string/number/bool values only
    )
    print(result.content, result.usage)
except RejectionError as e:
    print("blocked by UsageFlow:", e.message)

• Denied calls never reach the provider. A quota or policy denial raises RejectionError.
• Fails closed. If UsageFlow can't be reached, UsageFlowUnavailableError is raised and the provider is not called.
• Policies can reroute the call. If a matched policy tier names a model, the call runs on that model instead. result.model / result.provider show what actually ran, result.requested_model what you asked for, and result.vibe_policy the tier that fired.
• max_tokens defaults to 1024 and is enforced. For OpenAI reasoning models (o1, o3, o4, gpt-5) it is sent as max_completion_tokens, and temperature is dropped.
• tools and tool_choice are passed to the provider as-is; tool calls come back in result.tool_calls.
• Results have to_dict() for JSON responses.

A workflow that doesn't match an active policy is ignored. Usage is counted per identity, so all workflows for one identity share one total; each workflow only defines its own rules.

Stream

stream = client.stream(identity="cust_acme", model="gpt-4o-mini",
                       messages=[Message("user", "Tell me a story")])
for chunk in stream:
    print(chunk, end="", flush=True)
result = stream.result()  # usage is recorded when the stream ends

A denial raises when you call stream(), before any text is produced. Always drain the stream.

Embeddings (OpenAI)

result = client.embed(identity="cust_acme", model="text-embedding-3-small", input=["hello"])

Adjust a customer's balance

client.withdraw(identity="cust_acme", amount=50, idempotency_key="order-123", reason="export")
client.credit(identity="cust_acme", amount=50, idempotency_key="refund-123")  # reverses

idempotency_key is recorded with the event, but duplicates are not rejected yet — don't retry blindly.

Hold now, charge later

hold = client.withdraw_async(identity="cust_acme", amount=100, idempotency_key="job-7")
# ... do the work ...
client.close(hold.capture_id, 60)   # charge 60; omit the amount to charge the full 100

withdraw_async checks limits and policies up front; nothing is charged until close. A hold can be closed once. To close it from another process, pass identity= and the amount.

A hold stays open for 24 hours by default. Pass hold_for (a timedelta or seconds) to change it, and close before hold.expires_at (epoch ms): a hold that expires unclosed is released without charging. Closing early charges the amount you pass and releases the rest right away.

from datetime import timedelta
hold = client.withdraw_async(identity="cust_acme", amount=100, idempotency_key="job-8",
                             hold_for=timedelta(hours=2))

Configuration

The client connects on first use; call client.connect() at startup to surface a bad key early, and client.destroy() (or use it as a context manager) on shutdown. The client is thread-safe.

Not included yet

Image generation, speech, transcription, moderation and batch are not in the Python SDK. Async (asyncio) usage is not supported yet; call the client from a thread (e.g. asyncio.to_thread).

Example app

examples/vibe in the UsageFlow Python repository is a runnable HTTP app with the same routes as the JS and Go examples.