margint 0.1.0

Per-customer AI cost tracking and budget enforcement for AI-native SaaS. Track LLM spend per customer, enforce budgets, see margin in real time.

margint

Per-customer AI margin in three lines of code.

Tag every LLM call with a customer ID and Margint shows you which customers actually make you money — after AI costs. No proxy, no latency hit. Privacy-first: tokens, model, cost. Never your prompts.

pip install margint
# or: uv add margint  /  poetry add margint

import os
import openai
from margint import Margint

m = Margint(api_key=os.environ["MARGINT_API_KEY"])
client = m.wrap(openai.OpenAI(), customer_id=user.id, feature="chat")

client.chat.completions.create(model="gpt-4o", messages=messages)
# → tracked. Open the dashboard to see margin per customer.

Get a key at app.margint.dev. Free up to 100k events / month, GitHub OAuth, no credit card.

---

Three integration patterns

Pick whichever fits. Mix them.

1. wrap() — zero-touch

Wrap your LLM client once; every call is tracked.

import openai
from margint import Margint

m = Margint(api_key=os.environ["MARGINT_API_KEY"])
client = m.wrap(openai.OpenAI(), customer_id="cust_abc", feature="chat")

response = client.chat.completions.create(
    model="gpt-4o",
    messages=[{"role": "user", "content": "hi"}],
)

Works with OpenAI- and Anthropic-shaped responses (response.usage, response.model). For other providers, use track().

2. track() — manual

For custom HTTP, multi-customer requests, or providers without a native SDK.

response = my_llm_call()
m.track(
    customer_id=request.user.id,
    feature="summarize",
    provider="anthropic",
    model="claude-sonnet-4-20250514",
    input_tokens=response.usage.input_tokens,
    output_tokens=response.usage.output_tokens,
)

Cost is computed locally from the bundled pricing table — no network hop.

3. guarded_call() — kill switch

Block the call before it bills you when a customer is over their monthly budget.

from margint import BudgetExceededError

try:
    response = m.guarded_call(
        customer_id="cust_abc",
        feature="agent",
        fn=lambda: client.chat.completions.create(model="gpt-4o", messages=msgs),
    )
except BudgetExceededError as e:
    return {"error": "Budget exceeded", "breaches": [b.__dict__ for b in e.breaches]}, 402

Budget checks are cached for 60 s, so guarded_call stays fast in hot paths. It does not auto-track — combine with wrap() or track() if you want both.

---

Framework quickstarts

FastAPI

# app/margint_client.py
import os
from margint import Margint

margint = Margint(api_key=os.environ["MARGINT_API_KEY"])

# app/main.py
import openai
from contextlib import asynccontextmanager
from fastapi import FastAPI
from .margint_client import margint

@asynccontextmanager
async def lifespan(app: FastAPI):
    yield
    margint.shutdown()  # flush on exit

app = FastAPI(lifespan=lifespan)

@app.post("/chat")
def chat(user_id: str, messages: list[dict]):
    client = margint.wrap(openai.OpenAI(), customer_id=user_id, feature="chat")
    return client.chat.completions.create(model="gpt-4o", messages=messages)

Django

# myproject/apps.py
import os
from django.apps import AppConfig
from margint import Margint

margint: Margint | None = None

class CoreConfig(AppConfig):
    name = "core"

    def ready(self):
        global margint
        margint = Margint(api_key=os.environ["MARGINT_API_KEY"])

# core/views.py
import openai
from django.http import JsonResponse
from .apps import margint

def chat(request):
    client = margint.wrap(openai.OpenAI(), customer_id=request.user.id, feature="chat")
    response = client.chat.completions.create(model="gpt-4o", messages=request.POST["messages"])
    return JsonResponse(response.model_dump())

atexit handles flushing on process exit automatically.

Flask

import os
import openai
from flask import Flask, request, jsonify
from margint import Margint

app = Flask(__name__)
margint = Margint(api_key=os.environ["MARGINT_API_KEY"])

@app.post("/chat")
def chat():
    client = margint.wrap(openai.OpenAI(), customer_id=request.json["user_id"], feature="chat")
    response = client.chat.completions.create(model="gpt-4o", messages=request.json["messages"])
    return jsonify(response.model_dump())

---

Configuration

Margint(
    api_key: str,                                # required
    endpoint: str = "https://app.margint.dev/api/ingest/events",
    budget_endpoint: str = "https://app.margint.dev/api/budgets/check",
    flush_interval_seconds: float = 5.0,
    max_batch_size: int = 50,                    # flushes early when reached
    budget_cache_ttl_seconds: float = 60.0,
)

Self-hosting? Override endpoint and budget_endpoint.

---

Async

v0.1 is synchronous only. The flush worker runs on a background thread, so track() returns immediately and won't block async frameworks. A native AsyncMargint is on the roadmap — email [e-mail] if it matters for your stack.

---

Troubleshooting

Events not appearing?

• Verify api_key matches a key in Settings → API Keys.
• Call m.flush() — the 5 s timer may not have fired in short scripts.
• Check egress for app.margint.dev.

Cost shows as $0?

• Model isn't in the bundled pricing table. Pass cost_microdollars=... directly on track(), or email [e-mail] to add it.

wrap() not tracking?

• Response must expose .usage (with prompt_tokens/input_tokens) and .model. Otherwise call track() directly.

---

MIT licensed. © Margint. Questions: [e-mail].