obsidian/wiki/tech-patterns/cost-tracker-integration.md

title	tags	created	updated	updated3
AI Cost Tracker — Integrating a New Project	how-to ai cost-tracking integration	2026-04-27	2026-04-27	2026-04-27 (v3 — ContextVar pitfall, .env wipeout risk, docker inspect pattern)

Integrating a New Project with AI Cost Tracker

Step-by-step guide for connecting any Oliver backend project to the shared cost-tracker.

Service URLs

Environment	Base URL
Dev	https://optical-dev.oliver.solutions/cost-tracker/v1
Prod (future)	https://cost.oliver.agency/v1

Health check: GET {base_url}/health → {"status":"ok","db":"ok"}

Prerequisites

• Access to https://optical-dev.oliver.solutions/cost-tracker/ (Microsoft SSO or ask Vadym for dev access)
• Your project is a Python backend (FastAPI/other) — or adaptable for any HTTP client

---

Step 1 — Get an API key

1. Open the Admin UI → API Keys → Create key
2. Name it after your project (e.g. video-accessibility-dev)
3. Copy the key — it is shown only once
4. Store in your project's .env as COST_TRACKER_API_KEY

---

Step 2 — Add environment variables

COST_TRACKER_BASE_URL=https://optical-dev.oliver.solutions/cost-tracker/v1
COST_TRACKER_API_KEY=ct_live_xxxxxxxxxxxxxxxxxxxx
COST_TRACKER_SOURCE_APP=your-project-name

Gotcha — Docker doesn't re-read env_file on restart.
After editing .env, run docker compose up -d --force-recreate <service>.
Verify with:

docker inspect <container> 2>/dev/null | python3 -c \
  "import sys,json; cfg=json.load(sys.stdin)[0]['Config']['Env']; [print(e) for e in cfg if 'COST' in e]"

Gotcha — leading spaces in .env = key silently ignored.
When appending via SSH, use echo "KEY=val" >> .env — never heredoc over SSH (shell quoting adds spaces).

Gotcha — .env can be silently wiped during git pull or Docker rebuild.
Projects that bake source into the image (COPY . . in Dockerfile, no volume mount) will overwrite the server's .env on any fresh build or pull.
Always re-run grep COST_TRACKER .env AND the docker inspect command above after any redeploy.

---

Step 3 — Create a lightweight HTTP client

No SDK package exists — use httpx directly. Create core/cost_tracker.py:

import httpx
from app.core.config import settings

_client: httpx.AsyncClient | None = None

def get_client() -> httpx.AsyncClient:
    global _client
    if _client is None:
        _client = httpx.AsyncClient(
            base_url=settings.cost_tracker_base_url,
            headers={"X-API-Key": settings.cost_tracker_api_key},
            timeout=10.0,
        )
    return _client

Add to Settings in core/config.py:

cost_tracker_base_url: str = "https://optical-dev.oliver.solutions/cost-tracker/v1"
cost_tracker_api_key: str = ""
cost_tracker_source_app: str = "my-project"

---

Step 4 — Wrap AI calls (preflight → call → record)

This is the core pattern. Every paid AI call follows three steps.

Pitfall — do NOT use contextvars.ContextVar to pass user identity.
Python's ContextVar copies the context at task creation time. If your AI call is invoked via asyncio.wait_for(), asyncio.create_task(), or asyncio.ensure_future(), a ContextVar set in the calling coroutine will not be visible inside the spawned task — it silently returns the default value ("" in most cases), so user_external_id ends up empty in the cost tracker.
Always pass user_external_id: str as an explicit function parameter all the way down to the function that calls record().

from app.core.cost_tracker import get_client
import time

async def call_gemini_with_tracking(
    prompt: str,
    user_id: str,
    job_id: str,
    project_id: str | None = None,
) -> GenerateContentResponse:

    ct = get_client()

    # 1. Preflight — checks budget before making the AI call
    preflight = await ct.post("/preflight", json={
        "user_external_id": user_id,
        "project_external_id": project_id,
        "job_external_id": job_id,
        "source_app": settings.cost_tracker_source_app,
        "model": "gemini-2.5-pro-preview",
        "estimated_units": {
            "input_tokens": len(prompt) // 4,   # rough estimate
            "output_tokens": 2048,
        },
    })
    preflight.raise_for_status()
    pf = preflight.json()

    if not pf["allow"]:
        raise Exception(f"Budget exceeded: {pf.get('deny_reason')}")

    # 2. Actual AI call
    t0 = time.monotonic()
    response = await client.models.generate_content(
        model="gemini-2.5-pro-preview",
        contents=prompt,
    )
    elapsed_ms = int((time.monotonic() - t0) * 1000)

    # 3. Record actual usage
    await ct.post("/usage/record", json={
        "request_id": pf["request_id"],
        "user_external_id": user_id,
        "project_external_id": project_id,
        "job_external_id": job_id,
        "source_app": settings.cost_tracker_source_app,
        "model": "gemini-2.5-pro-preview",
        "units": {
            "input_tokens": response.usage_metadata.prompt_token_count,
            "output_tokens": response.usage_metadata.candidates_token_count,
        },
        "latency_ms": elapsed_ms,
        "status": "success",
    })

    return response

For providers without token metadata (ElevenLabs, Google Cloud TTS), use chars instead:

await ct.post("/usage/record", json={
    ...
    "model": "eleven_multilingual_v2",
    "units": {"chars": len(text)},
    "latency_ms": elapsed_ms,
    "status": "success",
})

See wiki/tech-patterns/cost-tracker-providers for all provider details.

---

Step 5 — Handle errors gracefully (don't break the AI pipeline)

Cost tracking should never crash your main pipeline. Wrap calls:

async def safe_preflight(ct, payload) -> dict:
    """Returns allow=True by default if cost-tracker is unreachable."""
    try:
        r = await ct.post("/preflight", json=payload)
        r.raise_for_status()
        return r.json()
    except Exception as e:
        logger.warning(f"Cost tracker preflight failed (allowing call): {e}")
        return {"allow": True, "request_id": None, "estimated_cost_usd": 0}

async def safe_record(ct, payload) -> None:
    try:
        r = await ct.post("/usage/record", json=payload)
        r.raise_for_status()
    except Exception as e:
        logger.warning(f"Cost tracker record failed (call already made): {e}")

Note: For higher reliability, implement a SQLite outbox — save failed records to a local DB and retry on next request. This is optional for most projects.

---

Step 6 — Sync users and projects (optional but recommended)

When a user is created or a project is configured in your app, mirror it to the cost-tracker so it appears in the analytics UI with a proper name:

# On user create/login — field is "external_id" (NOT "user_external_id")
await ct.post("/users/upsert", json={
    "external_id": str(user.id),   # ← correct field name
    "email": user.email,
    "full_name": user.full_name,
    # source_app is inferred from API key — do not send it
})

# On project create
await ct.post("/projects/upsert", json={
    "external_id": str(project.id),
    "source_app": settings.cost_tracker_source_app,
    "name": project.name,
    "workspace_id": "your-workspace-id",  # from Admin UI
})

---

Step 7 — Create Workspace / Team / Project in Admin UI

Before going live:

1. Admin UI → https://optical-dev.oliver.solutions/cost-tracker/
2. Workspaces → Create (e.g. "Ford", "H&M", or "Oliver Internal")
3. Teams → Create under workspace (e.g. "Video Production")
4. Projects → Create under team, set source_app = your project name
5. API Keys → Create key, assign to the project

Jobs sent before a project is created appear as Unassigned in the dashboard — assignable later in bulk.

---

Step 8 — Set budgets and alerts

Admin UI → Budgets → Create:

• scope_type: workspace / team / project
• amount_usd: monthly limit
• alert_thresholds: [0.5, 0.8, 1.0] → email alerts at 50%, 80%, 100%
• hard_limit: true → preflight returns allow=false when exceeded

---

Step 9 — Smoke test

BASE="https://optical-dev.oliver.solutions/cost-tracker/v1"
KEY="ct_live_your_key_here"

# 1. Health
curl "$BASE/health"
# → {"status":"ok","db":"ok"}

# 2. Preflight
curl -X POST "$BASE/preflight" \
  -H "X-API-Key: $KEY" \
  -H "Content-Type: application/json" \
  -d '{"user_external_id":"test-user","source_app":"my-project","model":"gemini-2.5-pro-preview","estimated_units":{"input_tokens":1000,"output_tokens":200}}'
# → {"allow":true,"estimated_cost_usd":...,"request_id":"..."}

# 3. Record
curl -X POST "$BASE/usage/record" \
  -H "X-API-Key: $KEY" \
  -H "Content-Type: application/json" \
  -d '{"request_id":"<from above>","user_external_id":"test-user","source_app":"my-project","model":"gemini-2.5-pro-preview","units":{"input_tokens":987,"output_tokens":180},"latency_ms":1200,"status":"success"}'
# → {"event_id":"...","cost_usd":0.00214}

Then open the Admin UI Dashboard — the test event should appear within seconds.

---

Troubleshooting

Symptom	Cause	Fix
401 Unauthorized	Wrong or missing API key	Check COST_TRACKER_API_KEY; verify key active in Admin UI
preflight.allow = false	Budget exceeded	Admin UI → Budgets → raise limit
Events missing from dashboard	source_app mismatch	Ensure source_app matches in preflight, record, and project upsert
cost_usd = null on events	Model not in pricing table	Admin UI → Pricing → add model manually, or wait for LiteLLM sync
502/504 from reverse proxy	cost-tracker containers down	ssh optical-dev "docker compose -f /opt/ai-cost-tracker/infra/docker-compose.yml ps"
Env vars not picked up after editing .env	docker compose restart doesn't re-read env_file	Use docker compose up -d --force-recreate <service>
Env vars in .env have leading spaces	Heredoc over SSH adds indentation	Use echo "KEY=val" >> .env per line; verify with grep COST_TRACKER .env
source_app shows in breakdown but "By Project" shows —	project_id null — not sent by integration	Use source_app dimension in Pivot Explorer instead of project_id
user_external_id is empty in all events	ContextVar not propagating across asyncio.wait_for / task boundary	Pass user_external_id as an explicit function parameter — do not use ContextVar
COST_TRACKER env vars disappear after deploy	git pull or Docker rebuild overwrites server .env	Re-add with echo commands; verify with docker inspect python3 one-liner after every rebuild

Related articles

• wiki/architecture/ai-cost-tracker — full system diagram and design decisions
• wiki/tech-patterns/cost-tracker-pricing-sources — pricing pipeline
• wiki/tech-patterns/cost-tracker-providers — billing units per AI provider