tokpulse 0.1.0

Lightweight LLM cost tracking for OpenAI, Anthropic, and Google AI

Tokpulse - Python SDK

Lightweight, non-blocking LLM cost tracking for OpenAI, Anthropic, and Google AI.

Installation

pip install tokpulse

# With provider-specific extras
pip install tokpulse[openai]
pip install tokpulse[anthropic]
pip install tokpulse[google]
pip install tokpulse[all]

Quick Start

from tokpulse import TokpulseTracker
from openai import OpenAI

# Initialize tracker
tracker = TokpulseTracker(
    api_key="lct_your_api_key",
    project="my-app",
    environment="production",
)

# Wrap your OpenAI client
client = tracker.openai(OpenAI())

# Use as normal - tracking is automatic
response = client.chat.completions.create(
    model="gpt-4o",
    messages=[{"role": "user", "content": "Hello!"}]
)

# Streaming works too
stream = client.chat.completions.create(
    model="gpt-4o",
    messages=[{"role": "user", "content": "Tell me a story"}],
    stream=True
)
for chunk in stream:
    print(chunk.choices[0].delta.content or "", end="")

# Flush before exit (important for serverless)
tracker.flush()

Provider Support

OpenAI

from openai import OpenAI

client = tracker.openai(OpenAI())
response = client.chat.completions.create(...)

Anthropic

from anthropic import Anthropic

client = tracker.anthropic(Anthropic())
message = client.messages.create(
    model="claude-sonnet-4-5-20250514",
    max_tokens=1024,
    messages=[{"role": "user", "content": "Hello!"}]
)

Google AI

import google.generativeai as genai

genai.configure(api_key="...")
model = tracker.google(genai.GenerativeModel("gemini-2.5-flash"))
response = model.generate_content("Hello!")

Configuration

tracker = TokpulseTracker(
    api_key="lct_xxx",             # Required
    api_url="https://...",         # API endpoint (default: api.tokpulse.io)
    allow_insecure_http=False,     # Set True only for http://localhost in local dev
    batch_size=10,                 # Trigger an early flush when this many events are queued
    flush_interval_seconds=5,      # Auto-flush interval (background thread)
    max_queue_size=1000,           # Max queued events; excess events are silently dropped
    project="my-app",              # Tag all events
    environment="production",      # Tag with environment
    debug=False,                   # Enable logging
    on_error=my_handler,           # Error callback: fn(error: Exception, event: UsageEvent)
    timeout_seconds=5,             # Request timeout
    max_retries=3,                 # Retry attempts
)

api_url must use https:// by default. Plain HTTP is only allowed for localhost (http://localhost, http://127.0.0.1, http://[::1]) when allow_insecure_http=True is explicitly set.

Context Manager

with TokpulseTracker(api_key="...") as tracker:
    client = tracker.openai(OpenAI())
    response = client.chat.completions.create(...)
# Auto-flush on exit

Manual Tracking

tracker.track(
    provider="openai",
    model="gpt-4o",
    input_tokens=100,
    output_tokens=50,
    cached_input_tokens=20,
)

track() validates inputs and raises ValueError when provider/model are empty or token counts are negative/non-integer.

Flush And Close

# Drain currently queued events
tracker.flush()

# Close: stop background workers, flush queue, and attempt pending retries once
tracker.close()

Retry backoff is scheduled in a background retry worker (non-blocking for the flush worker).

Serverless / Lambda

Always flush before the function exits:

def handler(event, context):
    response = client.chat.completions.create(...)
    tracker.flush()  # Important!
    return {"statusCode": 200, "body": response.choices[0].message.content}

Streaming Notes

Streaming usage is tracked when streams are fully consumed and also when iteration stops early (for example, break).

Async Clients

Full async wrappers are provided for OpenAI and Anthropic. Use tracker.openai_async() and tracker.anthropic_async() to get an async-native wrapper, then call await tracker.aflush() to drain the queue without blocking the event loop.

import asyncio
from openai import AsyncOpenAI
from anthropic import AsyncAnthropic
from tokpulse import TokpulseTracker

tracker = TokpulseTracker(api_key="lct_your_api_key")

# Async OpenAI
async def run_openai():
    client = tracker.openai_async(AsyncOpenAI())
    response = await client.chat.completions.create(
        model="gpt-4o",
        messages=[{"role": "user", "content": "Hello!"}],
    )
    await tracker.aflush()

# Async Anthropic
async def run_anthropic():
    client = tracker.anthropic_async(AsyncAnthropic())
    message = await client.messages.create(
        model="claude-sonnet-4-6",
        max_tokens=1024,
        messages=[{"role": "user", "content": "Hello!"}],
    )
    await tracker.aflush()

# Close without blocking the event loop
async def shutdown():
    await tracker.aclose()

Streaming works the same way — iterate with async for and usage is captured in the generator's finally block.

License

MIT