Stripe integration

Unimeter records and aggregates usage events in real time. Stripe handles invoicing and payments. Together they give you a complete usage-based billing pipeline without building aggregation logic yourself.

This guide walks through a working example: your application ingests usage events into Unimeter, and when Stripe creates an invoice it queries the usage total and adds a line item automatically.

How it fits together

The integration has two sides. On the recording side, your application sends usage events to Unimeter every time a billable action happens. Unimeter aggregates them in real time and keeps a running total per customer per billing period.

On the invoicing side, Stripe sends your server a webhook when it creates a new invoice. Your webhook handler takes the customer ID from the invoice, maps it to a Unimeter account, queries the usage total for the last completed billing period, and creates a line item on that invoice with the metered amount. Stripe then finalizes the invoice and charges the customer as usual.

Unimeter owns the usage data. Stripe owns the billing. Your webhook handler is the bridge between the two, and it only runs when Stripe asks for the numbers.

Prerequisites

A Stripe account with an API key and webhook signing secret.
A running Unimeter cluster (single-node is fine for development).
The Unimeter SDK installed:

Go
Python

go get github.com/unimeter/go-unimeter@latest

pip install unimeter-python

Step 1: Define your metric

Register a metric that counts API calls per calendar month. A BillingCycleDay of 1 means each period starts on the first of the month, matching the default Stripe subscription cycle.

Go
Python

client, _ := billing.New([]string{"localhost:7001"})

err := client.Metrics.Create(ctx, billing.MetricSchema{
    Code:            "api_calls",
    AggType:         billing.AggCount,
    PeriodType:      billing.PeriodCalendar,
    BillingCycleDay: 1,
})

from unimeter import AsyncClient, MetricSchema, AggType, PeriodType

async with AsyncClient(["localhost:7001"]) as client:
    await client.metrics.create(MetricSchema(
        code="api_calls",
        agg_type=AggType.COUNT,
        period_type=PeriodType.CALENDAR,
        billing_cycle_day=1,
    ))

You only need to do this once. The metric definition persists across restarts.

Step 2: Record usage

Call Ingest every time a billable action happens. A common pattern is an HTTP middleware that fires after each request.

Go
Python

func usageMiddleware(next http.Handler, client *billing.Client) http.Handler {
    return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
        next.ServeHTTP(w, r)

        accountID, _ := strconv.ParseUint(r.Header.Get("X-Account-ID"), 10, 64)
        if accountID == 0 {
            return
        }

        client.Ingest(r.Context(), []billing.Event{{
            AccountID:  accountID,
            MetricCode: "api_calls",
            Value:      1,
        }})
    })
}

from unimeter import Event

async def usage_middleware(request, client):
    """Call after each request to record one API call."""
    account_id = int(request.headers.get("X-Account-ID", "0"))
    if account_id == 0:
        return

    await client.ingest([
        Event(account_id=account_id, metric_code="api_calls", value=1),
    ])

Events are deduplicated by their idempotency key, so retries are safe.

Step 3: Handle the Stripe webhook

When Stripe finalizes an invoice it sends an invoice.created webhook. Your handler queries Unimeter for the customer’s usage during the last billing period and creates a line item on the invoice.

Go
Python

package main

import (
    "encoding/json"
    "fmt"
    "io"
    "log"
    "net/http"
    "os"
    "strconv"

    billing "github.com/unimeter/go-unimeter"
    "github.com/stripe/stripe-go/v82"
    "github.com/stripe/stripe-go/v82/invoiceitem"
    "github.com/stripe/stripe-go/v82/webhook"
)

// Map Stripe customer IDs to Unimeter account IDs.
// In production, store this in your database or in Stripe customer metadata.
var customerToAccount = map[string]uint64{
    "cus_demo_alice": 1,
    "cus_demo_bob":   2,
}

func main() {
    stripe.Key = os.Getenv("STRIPE_SECRET_KEY")
    webhookSecret := os.Getenv("STRIPE_WEBHOOK_SECRET")

    client, err := billing.New([]string{os.Getenv("BILLING_NODES")})
    if err != nil {
        log.Fatal(err)
    }
    defer client.Close()

    http.HandleFunc("/webhook", func(w http.ResponseWriter, r *http.Request) {
        body, _ := io.ReadAll(r.Body)
        event, err := webhook.ConstructEvent(
            body, r.Header.Get("Stripe-Signature"), webhookSecret,
        )
        if err != nil {
            http.Error(w, "invalid signature", http.StatusBadRequest)
            return
        }
        if event.Type != "invoice.created" {
            w.WriteHeader(http.StatusOK)
            return
        }

        var invoice stripe.Invoice
        json.Unmarshal(event.Data.Raw, &invoice)

        accountID, ok := customerToAccount[invoice.Customer.ID]
        if !ok {
            w.WriteHeader(http.StatusOK)
            return
        }

        // Query last completed billing period.
        usage, err := client.Query(r.Context(), billing.QueryRequest{
            AccountID:  accountID,
            MetricCode: "api_calls",
            Period:     billing.LastBillingPeriod(1),
        })
        if err != nil {
            http.Error(w, "query failed", http.StatusInternalServerError)
            return
        }

        // Create a line item on the Stripe invoice.
        period := billing.LastBillingPeriod(1)
        idempotencyKey := fmt.Sprintf(
            "unimeter-%d-api_calls-%s", accountID, period.Start.Format("2006-01"),
        )
        _, err = invoiceitem.New(&stripe.InvoiceItemParams{
            Customer:    stripe.String(invoice.Customer.ID),
            Invoice:     stripe.String(invoice.ID),
            Description: stripe.String("API calls"),
            Quantity:    stripe.Int64(int64(usage.Value.Count)),
            UnitAmount:  stripe.Int64(10), // $0.001 per call
            Currency:    stripe.String("usd"),
            Params: stripe.Params{
                IdempotencyKey: stripe.String(idempotencyKey),
            },
        })
        if err != nil {
            log.Printf("stripe error: %v", err)
        }
        w.WriteHeader(http.StatusOK)
    })

    log.Println("listening on :4242")
    log.Fatal(http.ListenAndServe(":4242", nil))
}

import json
import os

import stripe
from aiohttp import web

from unimeter import AsyncClient, last_billing_period

stripe.api_key = os.environ["STRIPE_SECRET_KEY"]
webhook_secret = os.environ["STRIPE_WEBHOOK_SECRET"]

CUSTOMER_TO_ACCOUNT = {
    "cus_demo_alice": 1,
    "cus_demo_bob": 2,
}

client: AsyncClient  # initialized at startup


async def webhook_handler(request: web.Request) -> web.Response:
    body = await request.read()
    sig = request.headers.get("Stripe-Signature", "")

    try:
        event = stripe.Webhook.construct_event(body, sig, webhook_secret)
    except stripe.error.SignatureVerificationError:
        return web.Response(status=400, text="invalid signature")

    if event["type"] != "invoice.created":
        return web.Response(status=200)

    invoice = event["data"]["object"]
    customer_id = invoice["customer"]
    account_id = CUSTOMER_TO_ACCOUNT.get(customer_id)
    if account_id is None:
        return web.Response(status=200)

    # Query last completed billing period.
    period = last_billing_period(1)
    usage = await client.query(
        account_id, "api_calls", period,
    )

    # Create a line item on the Stripe invoice.
    idempotency_key = f"unimeter-{account_id}-api_calls-{period.start:%Y-%m}"
    stripe.InvoiceItem.create(
        customer=customer_id,
        invoice=invoice["id"],
        description="API calls",
        quantity=usage.value.count,
        unit_amount=10,  # $0.001 per call
        currency="usd",
        idempotency_key=idempotency_key,
    )
    return web.Response(status=200)


async def on_startup(app: web.Application) -> None:
    global client
    nodes = os.environ["BILLING_NODES"].split(",")
    client = AsyncClient(nodes)
    await client.connect()


async def on_cleanup(app: web.Application) -> None:
    await client.close()


app = web.Application()
app.on_startup.append(on_startup)
app.on_cleanup.append(on_cleanup)
app.router.add_post("/webhook", webhook_handler)

if __name__ == "__main__":
    web.run_app(app, port=4242)

The important mapping is between Stripe customer IDs and Unimeter account IDs. In the example this is a hardcoded map; in production, store the Unimeter account ID in Stripe customer metadata or in your database.

The idempotency key includes the account, metric, and period start date. If Stripe retries the webhook, the invoice item creation is skipped because Stripe has already seen that key.

Step 4: Run and test

Set the required environment variables and start the server:

Go
Python

export STRIPE_SECRET_KEY="sk_test_..."
export STRIPE_WEBHOOK_SECRET="whsec_..."
export BILLING_NODES="localhost:7001"
go run main.go

export STRIPE_SECRET_KEY="sk_test_..."
export STRIPE_WEBHOOK_SECRET="whsec_..."
export BILLING_NODES="localhost:7001"
python main.py

In a separate terminal, forward Stripe test webhooks to your local server using the Stripe CLI:

stripe listen --forward-to localhost:4242/webhook

Create a test invoice in the Stripe dashboard or via the API and watch the webhook handler query Unimeter and attach a line item.

What’s next

Go SDK reference — full API documentation for the Go client.
Python SDK reference — full API documentation for the Python client.
Events and metrics — how events are recorded and aggregated.
Metric types — COUNT, SUM, MAX, LATEST, and COUNT UNIQUE.
Alerts — get notified when usage crosses a threshold.