Metadata-Version: 2.4
Name: ai-token-tracker
Version: 0.2.3
Summary: Python SDK for capturing LLM traffic and ingesting envelopes into Ai Token Tracker
Author: Ai Token Tracker
License: MIT
Project-URL: Homepage, https://github.com/ai-token-tracker/ai-token-tracker
Project-URL: Repository, https://github.com/ai-token-tracker/ai-token-tracker
Project-URL: Documentation, https://github.com/ai-token-tracker/ai-token-tracker/tree/main/sdks/python
Project-URL: Issues, https://github.com/ai-token-tracker/ai-token-tracker/issues
Keywords: llm,observability,openai,anthropic,sdk
Classifier: Development Status :: 4 - Beta
Classifier: Intended Audience :: Developers
Classifier: License :: OSI Approved :: MIT License
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3.10
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Classifier: Programming Language :: Python :: 3.13
Classifier: Topic :: Software Development :: Libraries :: Python Modules
Classifier: Typing :: Typed
Requires-Python: >=3.10
Description-Content-Type: text/markdown
Provides-Extra: requests
Requires-Dist: requests>=2.31; extra == "requests"
Provides-Extra: httpx
Requires-Dist: httpx>=0.27; extra == "httpx"
Provides-Extra: all
Requires-Dist: requests>=2.31; extra == "all"
Requires-Dist: httpx>=0.27; extra == "all"
Provides-Extra: dev
Requires-Dist: pytest>=8.2; extra == "dev"
Requires-Dist: requests>=2.31; extra == "dev"
Requires-Dist: httpx>=0.27; extra == "dev"

# Ai Token Tracker Python SDK

`ai-token-tracker` captures outbound LLM traffic and posts ingest envelopes to Ai Token Tracker.

Supports:

1. Automatic HTTP interception (`requests`, `httpx` sync/async)
2. SDK wrapper integration (`begin_llm_call` scope)
3. Diagnostics fallback (metadata-only safety net)

## Requirements

- Python 3.10+

## Installation

```bash
pip install ai-token-tracker
```

Optional extras:

```bash
pip install ai-token-tracker[requests]
pip install ai-token-tracker[httpx]
pip install ai-token-tracker[all]
```

## Defaults

- `enable_auto_interception=True`
- `enable_diagnostics_fallback=True`

## Quick Start (Plain Script)

```python
from ai_token_tracker import (
    AiTokenTrackerIngestionClient,
    AiTokenTrackerOptions,
    AiTokenTrackerSdkClient,
)

ingestion = AiTokenTrackerIngestionClient(
    AiTokenTrackerOptions(auth_token="atk_your_key")
)

sdk = AiTokenTrackerSdkClient(ingestion)

scope = sdk.begin_llm_call(
    request={"model": "gpt-4.1-mini", "input": "hello"},
    custom_filters={"workflow": "quickstart"},
    provider_hint="anthropic",
)

try:
    provider_response = {"id": "resp_123", "output": [{"type": "output_text", "text": "hi"}]}
    scope.complete(provider_response, status_code=200)
except Exception as ex:
    scope.fail(ex, status_code=500)
    raise
finally:
    ingestion.close()
```

## Automatic HTTP Interception

Interception is explicit install + option-gated:

```python
import requests
from ai_token_tracker import (
    AiTokenTrackerIngestionClient,
    AiTokenTrackerOptions,
    install_http_interception,
)

client = AiTokenTrackerIngestionClient(
    AiTokenTrackerOptions(
        auth_token="atk_your_key",
        enable_auto_interception=True,
    )
)

install_http_interception(client)

# Classified request is captured and ingested automatically.
requests.post(
    "https://api.openai.com/v1/responses",
    headers={"Authorization": "Bearer sk_test"},
    json={"model": "gpt-4.1-mini", "input": "hello"},
    timeout=10,
)

client.close()  # unpatches interceptors + shuts down background worker
```

### Scoped Custom Filters For Automatic Interception

Use interception scope to attach custom filters without wrapper integration:

```python
import requests
from ai_token_tracker import (
    AiTokenTrackerIngestionClient,
    AiTokenTrackerInterceptionScope,
    AiTokenTrackerOptions,
    install_http_interception,
)

client = AiTokenTrackerIngestionClient(
    AiTokenTrackerOptions(auth_token="atk_your_key", enable_auto_interception=True)
)
install_http_interception(client)

scope = AiTokenTrackerInterceptionScope.create({"workflow": "social_post_generation"})
scope.add_custom_filters({"job_id": "job-42", "customer_id": "cust-19"})

scope.open()
try:
    requests.post(
        "https://api.openai.com/v1/responses",
        headers={"Authorization": "Bearer sk_test"},
        json={"model": "gpt-4.1-mini", "input": "hello"},
        timeout=10,
    )
finally:
    scope.close()
```

Always close scope in `finally` to avoid leaking filters to later calls.

## SDK Wrapper Example

```python
from ai_token_tracker import AiTokenTrackerIngestionClient, AiTokenTrackerOptions, AiTokenTrackerSdkClient

ingestion = AiTokenTrackerIngestionClient(AiTokenTrackerOptions(auth_token="atk_your_key"))
tracker = AiTokenTrackerSdkClient(ingestion)
# method defaults to "POST" when omitted

scope = tracker.begin_llm_call(
    request={"model": "claude-sonnet-4-6", "max_tokens": 256, "messages": [{"role": "user", "content": "hello"}]},
    custom_filters={"tenant": "acme", "job": "summarization"},
    provider_hint="anthropic",
)

scope.complete({"id": "msg_123"}, status_code=200)
ingestion.close()
```

## Diagnostics Fallback Setup

Diagnostics fallback is best-effort metadata capture for uncovered traffic.

```python
from ai_token_tracker import AiTokenTrackerIngestionClient, AiTokenTrackerOptions, install_http_interception

client = AiTokenTrackerIngestionClient(
    AiTokenTrackerOptions(
        auth_token="atk_your_key",
        enable_diagnostics_fallback=True,
    )
)
install_http_interception(client)
```

Fallback behavior:

- captures method, url, status, headers
- does not guarantee request/response bodies
- intended for visibility and coverage-gap detection, not full fidelity auditing

## FastAPI Example

```python
from contextlib import asynccontextmanager
from fastapi import FastAPI
from ai_token_tracker import (
    AiTokenTrackerIngestionClient,
    AiTokenTrackerOptions,
    AiTokenTrackerSdkClient,
    install_http_interception,
)

ingestion = AiTokenTrackerIngestionClient(
    AiTokenTrackerOptions(auth_token="atk_your_key", enable_auto_interception=True)
)
tracker = AiTokenTrackerSdkClient(ingestion)

@asynccontextmanager
async def lifespan(app: FastAPI):
    install_http_interception(ingestion)
    yield
    ingestion.close()

app = FastAPI(lifespan=lifespan)

@app.get("/health")
def health() -> dict[str, str]:
    return {"status": "ok"}
```

## Background Worker / Job Pipeline Example

```python
from ai_token_tracker import AiTokenTrackerIngestionClient, AiTokenTrackerOptions, AiTokenTrackerSdkClient

ingestion = AiTokenTrackerIngestionClient(AiTokenTrackerOptions(auth_token="atk_your_key"))
tracker = AiTokenTrackerSdkClient(ingestion)

def run_job(job_id: str, prompt: str) -> None:
    scope = tracker.begin_llm_call(
        request={"model": "gpt-4.1-mini", "input": prompt},
        custom_filters={"job_id": job_id, "pipeline": "nightly"},
        provider_hint="anthropic",
    )
    try:
        response = {"id": f"resp_{job_id}", "output": [{"type": "output_text", "text": "done"}]}
        scope.complete(response)
    except Exception as ex:
        scope.fail(ex)
        raise

run_job("42", "summarize daily usage")
ingestion.close()
```

## Custom Filters Guidance

`custom_filters` is key/value metadata attached to each envelope:

- keep keys stable (`tenant`, `workflow`, `job_id`, `environment`)
- keep values low-cardinality when possible
- avoid putting secrets or full prompt bodies in filters

## Verification Checklist

1. Create client with valid `auth_token`.
2. For interception, set option flag and call `install_http_interception(client)`.
3. Send known LLM request (`/v1/responses` or `/chat/completions`).
4. Confirm event appears in AI Token Tracker dashboard.
5. Confirm app flow still works if ingest is unavailable (tracking paths are non-throwing).

## Public API Summary

- `AiTokenTrackerOptions(auth_token, enable_auto_interception=True, enable_diagnostics_fallback=True)`
- `AiTokenTrackerIngestionClient.track(...) -> TrackResult`
- `AiTokenTrackerIngestionClient.track_async(...) -> TrackResult`
- `AiTokenTrackerIngestionClient.close()`
- `AiTokenTrackerInterceptionScope.create(...)`, `open()`, `close()`
- `AiTokenTrackerSdkClient.begin_llm_call(...) -> CallScope`
- `CallScope.complete(...)`, `CallScope.fail(...)`
- `CallScope.complete_async(...)`, `CallScope.fail_async(...)`
- `install_http_interception(client)`, `uninstall_http_interception(client)`
