How to set a hard spending limit on the OpenAI API (per project, key or user)

import OpenAI, { APIError } from "openai";

// No SDK retries: it would retry the spend-limit 429 below too (see the 429 guide).
const client = new OpenAI({ maxRetries: 0 });
const MODEL = "gpt-5.4-mini";
// USD per 1M tokens. Use the rates your own contract carries.
const RATES = { input: 0.75, output: 4.5 };

const costUsd = (inputTokens: number, outputTokens: number) =>
  (inputTokens * RATES.input + outputTokens * RATES.output) / 1_000_000;

export class CapReached extends Error {}

/**
 * Spend per customer per UTC day. This keeps the ledger in memory, which is
 * correct for one process only: with several servers, keep it in Redis or
 * Postgres and make reserve() a single atomic increment-and-check.
 */
export class DailyCap {
  private spent = new Map<string, number>();
  private readonly limitUsd: number;

  constructor(limitUsd: number) {
    this.limitUsd = limitUsd;
  }

  private key(customer: string) {
    return `${customer}:${new Date().toISOString().slice(0, 10)}`;
  }

  reserve(customer: string, usd: number) {
    const used = this.spent.get(this.key(customer)) ?? 0;
    if (used + usd > this.limitUsd) {
      throw new CapReached(`not sent: ${customer} has used $${used.toFixed(4)} of its $${this.limitUsd} daily cap`);
    }
    this.spent.set(this.key(customer), used + usd);
  }

  // Replace the reservation with what the call actually cost.
  settle(customer: string, reservedUsd: number, actualUsd: number) {
    const used = this.spent.get(this.key(customer)) ?? 0;
    this.spent.set(this.key(customer), Math.max(0, used - reservedUsd + actualUsd));
  }
}

export async function answerFor(customer: string, prompt: string, cap: DailyCap, maxTokens = 500) {
  // Reserve the most this call can cost BEFORE making it, so ten concurrent
  // calls cannot all pass a check that only one of them fits under.
  // Two characters per token overestimates the prompt, on purpose.
  const reserved = costUsd(Math.ceil(prompt.length / 2), maxTokens);
  cap.reserve(customer, reserved);
  try {
    const res = await client.chat.completions.create({
      model: MODEL,
      max_completion_tokens: maxTokens,
      messages: [{ role: "user", content: prompt }],
    });
    cap.settle(customer, reserved, costUsd(res.usage?.prompt_tokens ?? 0, res.usage?.completion_tokens ?? 0));
    return res.choices[0]?.message.content ?? "";
  } catch (err) {
    cap.settle(customer, reserved, 0);
    // OpenAI's own project hard limit: stop, do not retry, tell a human.
    if (err instanceof APIError && err.code === "project_spend_limit_exceeded") {
      throw new CapReached("OpenAI project hard limit reached");
    }
    throw err;
  }
}

const cap = new DailyCap(0.005);
for (let i = 1; i <= 3; i++) {
  try {
    await answerFor("cus_123", "Write a one-line welcome for a new user.", cap);
    console.log(`call ${i}: served`);
  } catch (err) {
    if (!(err instanceof CapReached)) throw err;
    console.log(`call ${i}: refused, ${err.message}`);
  }
}

import datetime
import math
import threading

import openai
from openai import OpenAI

# No SDK retries: it would retry the spend-limit 429 below too (see the 429 guide).
client = OpenAI(max_retries=0)
MODEL = "gpt-5.4-mini"
# USD per 1M tokens. Use the rates your own contract carries.
RATES = {"input": 0.75, "output": 4.5}


def cost_usd(input_tokens: int, output_tokens: int) -> float:
    return (input_tokens * RATES["input"] + output_tokens * RATES["output"]) / 1_000_000


class CapReached(Exception):
    pass


class DailyCap:
    """Spend per customer per UTC day.

    This keeps the ledger in memory, which is correct for one process only:
    with several servers, keep it in Redis or Postgres and make reserve() a
    single atomic increment-and-check.
    """

    def __init__(self, limit_usd: float):
        self.limit_usd = limit_usd
        self.spent: dict[str, float] = {}
        self.lock = threading.Lock()

    def _key(self, customer: str) -> str:
        return f"{customer}:{datetime.datetime.now(datetime.timezone.utc).date()}"

    def reserve(self, customer: str, usd: float) -> None:
        with self.lock:
            used = self.spent.get(self._key(customer), 0.0)
            if used + usd > self.limit_usd:
                raise CapReached(f"not sent: {customer} has used ${used:.4f} of its ${self.limit_usd} daily cap")
            self.spent[self._key(customer)] = used + usd

    def settle(self, customer: str, reserved_usd: float, actual_usd: float) -> None:
        """Replace the reservation with what the call actually cost."""
        with self.lock:
            used = self.spent.get(self._key(customer), 0.0)
            self.spent[self._key(customer)] = max(0.0, used - reserved_usd + actual_usd)


def answer_for(customer: str, prompt: str, cap: DailyCap, max_tokens: int = 500) -> str:
    # Reserve the most this call can cost BEFORE making it, so ten concurrent
    # calls cannot all pass a check that only one of them fits under.
    # Two characters per token overestimates the prompt, on purpose.
    reserved = cost_usd(math.ceil(len(prompt) / 2), max_tokens)
    cap.reserve(customer, reserved)
    try:
        res = client.chat.completions.create(
            model=MODEL,
            max_completion_tokens=max_tokens,
            messages=[{"role": "user", "content": prompt}],
        )
    except openai.APIStatusError as err:
        cap.settle(customer, reserved, 0.0)
        # OpenAI's own project hard limit: stop, do not retry, tell a human.
        if err.code == "project_spend_limit_exceeded":
            raise CapReached("OpenAI project hard limit reached") from err
        raise
    except Exception:
        cap.settle(customer, reserved, 0.0)
        raise
    usage = res.usage
    cap.settle(customer, reserved, cost_usd(usage.prompt_tokens, usage.completion_tokens) if usage else 0.0)
    return res.choices[0].message.content or ""


if __name__ == "__main__":
    cap = DailyCap(0.005)
    for i in range(1, 4):
        try:
            answer_for("cus_123", "Write a one-line welcome for a new user.", cap)
            print(f"call {i}: served")
        except CapReached as err:
            print(f"call {i}: refused, {err}")