a2a-llm-tracker 0.0.10

A short description of your package

a2a-llm-tracker

Track LLM usage and costs across providers (OpenAI, Gemini, Anthropic, etc.) from a single place.

Installation

pip install a2a-llm-tracker

Quick Start (Recommended Pattern)

For applications making multiple LLM calls, use a singleton pattern to initialize once and reuse everywhere.

Step 1: Create a tracking module

Create tracking.py in your project:

# tracking.py
import os
import asyncio
import concurrent.futures
from a2a_llm_tracker import init

_meter = None

def get_meter():
    """Get or initialize the global meter singleton."""
    global _meter
    if _meter is None:
        try:
            client_id = os.getenv("CLIENT_ID", "")
            client_secret = os.getenv("CLIENT_SECRET", "")

            with concurrent.futures.ThreadPoolExecutor() as executor:
                future = executor.submit(
                    asyncio.run,
                    init(client_id, client_secret, "my-app")
                )
                _meter = future.result(timeout=5)

        except Exception as e:
            print(f"LLM tracking initialization failed: {e}")
            return None
    return _meter

Step 2: Use it anywhere

from openai import OpenAI
from a2a_llm_tracker import analyze_response, ResponseType
from tracking import get_meter

def call_openai(prompt: str):
    client = OpenAI()
    response = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{"role": "user", "content": prompt}],
    )

    # Track usage
    meter = get_meter()
    if meter:
        analyze_response(response, ResponseType.OPENAI, meter)

    return response

Environment Variables

Set your CCS credentials:

export CLIENT_ID=your_client_id
export CLIENT_SECRET=your_client_secret
export OPENAI_API_KEY=sk-xxxxx

Query Total Usage & Costs

Retrieve your accumulated costs and token usage from CCS:

import os
import asyncio
from a2a_llm_tracker import init
from a2a_llm_tracker.sources import CCSSource

async def get_total_usage():
    client_id = os.getenv("CLIENT_ID")
    client_secret = os.getenv("CLIENT_SECRET")

    await init(
        client_id=client_id,
        client_secret=client_secret,
        application_name="my-app",
    )

    source = CCSSource(int(client_id))
    total_cost = await source.count_cost()
    total_tokens = await source.count_total_tokens()

    print(f"Total cost: ${total_cost:.4f}")
    print(f"Total tokens: {total_tokens}")

asyncio.run(get_total_usage())

Supported Providers

Provider	ResponseType
OpenAI	ResponseType.OPENAI
Google Gemini	ResponseType.GEMINI
Anthropic	ResponseType.ANTHROPIC
Cohere	ResponseType.COHERE
Mistral	ResponseType.MISTRAL
Groq	ResponseType.GROQ
Together AI	ResponseType.TOGETHER
AWS Bedrock	ResponseType.BEDROCK
Google Vertex AI	ResponseType.VERTEX

Documentation

• LiteLLM Wrapper - Auto-tracking via LiteLLM
• CCS Integration - Centralized tracking setup
• Response Analysis - Direct SDK tracking
• Pricing - Custom pricing configuration
• Building - Development and publishing

What This Package Does NOT Do

• Guess exact billing from raw text
• Replace provider SDKs
• Upload data anywhere automatically
• Require a backend or SaaS