AI budget control
AI providers bill per token. Without per-user limits, a single user can exhaust your entire monthly budget – through prompt attacks, runaway loops, or just heavy legitimate use.
A token bucket rate limit maps directly onto how AI billing works. You estimate the cost of each request in tokens, deduct it from the user’s bucket, and deny requests when the bucket is empty. The bucket refills over time, giving each user a sustained allowance without sharp rate-limit cliffs.
Alternatively, you can use a fixed window or sliding window limit to enforce a hard cap on spend per user per day, week, or month. For details on different approaches, see the rate limiting algorithms reference.
Arcjet handles bucket state across all instances of your application – no Redis or external state store required.
Get started
Section titled “Get started”This example deducts an estimated token cost from a per-user bucket before a tool runs. When the bucket is empty, the tool does not execute.
We assume you already have a Claude Agent SDK project set up. For helper options and denial behavior, see the Claude Agent SDK agent guard.
Install the dependencies:
# Export your Arcjet API key from https://app.arcjet.comexport ARCJET_KEY="ajkey_..."
npm install @arcjet/guard @anthropic-ai/claude-agent-sdkCreate the example:
import { tool } from "@anthropic-ai/claude-agent-sdk";import { launchArcjet, tokenBucket } from "@arcjet/guard";import { guardTool } from "@arcjet/guard/claude-agent-sdk/v0";import { z } from "zod";
const arcjet = launchArcjet({ key: process.env.ARCJET_KEY! });
const tokenBudget = tokenBucket({ bucket: "ai-tokens", refillRate: 2000, intervalSeconds: 3600, maxTokens: 5000,});
export const completePrompt = guardTool( arcjet, tool( "complete_prompt", "Complete a user prompt", { prompt: z.string(), estimatedTokens: z.number() }, async ({ prompt }) => ({ content: [{ type: "text", text: prompt }], }), ), { action: "prompt.completed", rules: (input) => [ tokenBudget({ key: "user123", // Replace with your authenticated user ID requested: Math.max(1, Math.ceil(input.estimatedTokens)), }), ], },);Then start or invoke the agent with a test prompt.
Requests appear in your Arcjet dashboard in real time.
This example deducts an estimated token cost from a per-user bucket before a tool runs. When the bucket is empty, the tool does not execute.
We assume you already have a LangChain project set up. For helper options and denial behavior, see the LangChain agent guard.
Install the dependencies:
# Export your Arcjet API key from https://app.arcjet.comexport ARCJET_KEY="ajkey_..."export ARCJET_ENV=development
pip install "arcjet[langchain-agents]" langchain langchain-openaiCreate the example:
import os
from arcjet.guard import TokenBucket, launch_arcjetfrom arcjet.guard.langchain import guard_toolfrom langchain_core.tools import tool
arcjet = launch_arcjet(key=os.environ["ARCJET_KEY"])token_budget = TokenBucket( refill_rate=2000, interval_seconds=3600, max_tokens=5000, bucket="ai-tokens",)
@toolasync def complete_prompt(prompt: str, estimated_tokens: int) -> dict: """Complete a user prompt.""" return {"prompt": prompt}
complete_prompt = guard_tool( guard=arcjet, tool=complete_prompt, action="prompt.completed", rules=lambda arguments, _config: [ token_budget( key="user123", # Replace with your authenticated user ID requested=max(1, int(arguments["estimated_tokens"])), ) ],)Then start or invoke the agent with a test prompt.
Requests appear in your Arcjet dashboard in real time.
This example deducts an estimated token cost from a per-user bucket before a tool runs. When the bucket is empty, the tool does not execute.
We assume you already have a Vercel AI SDK project set up. For helper options and denial behavior, see the Vercel AI SDK agent guard.
Install the dependencies:
# Export your Arcjet API key from https://app.arcjet.comexport ARCJET_KEY="ajkey_..."
npm install @arcjet/guard ai @ai-sdk/provider-utilsCreate the example:
import { launchArcjet, tokenBucket } from "@arcjet/guard";import { guardTool } from "@arcjet/guard/vercel-ai/v7";import { tool } from "ai";import { z } from "zod";
const arcjet = launchArcjet({ key: process.env.ARCJET_KEY! });
const tokenBudget = tokenBucket({ bucket: "ai-tokens", refillRate: 2000, intervalSeconds: 3600, maxTokens: 5000,});
export const completePrompt = guardTool( arcjet, tool({ description: "Complete a user prompt", inputSchema: z.object({ prompt: z.string(), estimatedTokens: z.number() }), execute: async ({ prompt }) => ({ prompt }), }), { action: "prompt.completed", actor: "user123", // Replace with your authenticated user ID rules: (input) => [ tokenBudget({ key: "user123", // Replace with your authenticated user ID requested: Math.max(1, Math.ceil(input.estimatedTokens)), }), ], },);Then start or invoke the agent with a test prompt.
Requests appear in your Arcjet dashboard in real time.
This example deducts an estimated token cost from a per-user bucket before a tool runs. When the bucket is empty, the tool does not execute.
We assume you already have a Genkit project set up. For helper options and denial behavior, see the Genkit agent guard.
Until @arcjet/guard/genkit/v1 is published, the import lives on
4e416787,
not in the published @arcjet/guard package on npm.
Install the dependencies:
# Export your Arcjet API key from https://console.arcjet.comexport ARCJET_KEY="ajkey_..."
npm install @arcjet/guard genkitCreate the example:
import { launchArcjet, tokenBucket } from "@arcjet/guard";import { guardTool } from "@arcjet/guard/genkit/v1";import { genkit, z } from "genkit";
const arcjet = launchArcjet({ key: process.env.ARCJET_KEY! });const ai = genkit({ // Configure your model plugin.});
const tokenBudget = tokenBucket({ bucket: "ai-tokens", refillRate: 2000, intervalSeconds: 3600, maxTokens: 5000,});
export const completePrompt = guardTool( arcjet, ai.defineTool( { name: "complete_prompt", description: "Complete a user prompt", inputSchema: z.object({ prompt: z.string(), estimatedTokens: z.number(), }), }, async ({ prompt }) => ({ prompt }), ), { action: "prompt.completed", rules: (input) => [ tokenBudget({ key: "user123", // Replace with your authenticated user ID requested: Math.max(1, Math.ceil(input.estimatedTokens)), }), ], },);Then start or invoke the agent with a test prompt.
Requests appear in your Arcjet dashboard in real time.
This example deducts an estimated token cost from a per-user bucket before a tool runs. When the bucket is empty, the tool does not execute.
We assume you already have a LangGraph project set up. For helper options and denial behavior, see the LangGraph agent guard.
Install the dependencies:
# Export your Arcjet API key from https://app.arcjet.comexport ARCJET_KEY="ajkey_..."
npm install @arcjet/guard @langchain/langgraph @langchain/coreCreate the example:
import { launchArcjet, tokenBucket } from "@arcjet/guard";import { guardTool } from "@arcjet/guard/langgraph/v1";import { tool } from "@langchain/core/tools";import { z } from "zod";
const arcjet = launchArcjet({ key: process.env.ARCJET_KEY! });
const tokenBudget = tokenBucket({ bucket: "ai-tokens", refillRate: 2000, intervalSeconds: 3600, maxTokens: 5000,});
export const completePrompt = guardTool( arcjet, tool( async ({ prompt }) => ({ prompt }), { name: "complete_prompt", description: "Complete a user prompt", schema: z.object({ prompt: z.string(), estimatedTokens: z.number(), }), }, ), { action: "prompt.completed", rules: (input) => [ tokenBudget({ key: "user123", // Replace with your authenticated user ID requested: Math.max(1, Math.ceil(input.estimatedTokens)), }), ], },);Then start or invoke the agent with a test prompt.
Requests appear in your Arcjet dashboard in real time.
This example deducts an estimated token cost from a per-user bucket before a tool runs. When the bucket is empty, the tool does not execute.
We assume you already have a OpenAI Agents project set up. For helper options and denial behavior, see the OpenAI Agents agent guard.
Install the dependencies:
# Export your Arcjet API key from https://app.arcjet.comexport ARCJET_KEY="ajkey_..."
npm install @arcjet/guard @openai/agentsCreate the example:
import { launchArcjet, tokenBucket } from "@arcjet/guard";import { guardTool } from "@arcjet/guard/openai-agents/v0";import { tool } from "@openai/agents";import { z } from "zod";
const arcjet = launchArcjet({ key: process.env.ARCJET_KEY! });
const tokenBudget = tokenBucket({ bucket: "ai-tokens", refillRate: 2000, intervalSeconds: 3600, maxTokens: 5000,});
export const completePrompt = guardTool( arcjet, tool({ name: "complete_prompt", description: "Complete a user prompt", parameters: z.object({ prompt: z.string(), estimatedTokens: z.number(), }), execute: async ({ prompt }) => ({ prompt }), }), { action: "prompt.completed", rules: (input) => [ tokenBudget({ key: "user123", // Replace with your authenticated user ID requested: Math.max(1, Math.ceil(input.estimatedTokens)), }), ], },);Then start or invoke the agent with a test prompt.
Requests appear in your Arcjet dashboard in real time.
This example deducts an estimated token cost from a per-user bucket before a tool runs. When the bucket is empty, the tool does not execute.
We assume you already have a Mastra project set up. For helper options and denial behavior, see the Mastra agent guard.
Install the dependencies:
# Export your Arcjet API key from https://app.arcjet.comexport ARCJET_KEY="ajkey_..."
npm install @arcjet/guard @mastra/coreCreate the example:
import { launchArcjet, tokenBucket } from "@arcjet/guard";import { guardTool } from "@arcjet/guard/mastra/v1";import { createTool } from "@mastra/core/tools";import { z } from "zod";
const arcjet = launchArcjet({ key: process.env.ARCJET_KEY! });
const tokenBudget = tokenBucket({ bucket: "ai-tokens", refillRate: 2000, intervalSeconds: 3600, maxTokens: 5000,});
export const completePrompt = guardTool( arcjet, createTool({ id: "complete-prompt", description: "Complete a user prompt", inputSchema: z.object({ prompt: z.string(), estimatedTokens: z.number(), }), async execute({ prompt }) { return { prompt }; }, }), { action: "prompt.completed", rules: (input) => [ tokenBudget({ key: "user123", // Replace with your authenticated user ID requested: Math.max(1, Math.ceil(input.estimatedTokens)), }), ], },);Then start or invoke the agent with a test prompt.
Requests appear in your Arcjet dashboard in real time.
This example deducts an estimated token cost from a per-user bucket before a tool runs. When the bucket is empty, the tool does not execute.
We assume you already have a Vercel Eve project set up. For helper options and denial behavior, see the Vercel Eve agent guard.
Install the dependencies:
# Export your Arcjet API key from https://app.arcjet.comexport ARCJET_KEY="ajkey_..."
npm install @arcjet/guard eveCreate the example:
import { launchArcjet, tokenBucket } from "@arcjet/guard";import { guardTool } from "@arcjet/guard/vercel-eve/v0";import { defineTool } from "eve/tools";import { z } from "zod";
const arcjet = launchArcjet({ key: process.env.ARCJET_KEY! });
const tokenBudget = tokenBucket({ bucket: "ai-tokens", refillRate: 2000, intervalSeconds: 3600, maxTokens: 5000,});
export default guardTool( arcjet, defineTool({ description: "Complete a user prompt", inputSchema: z.object({ prompt: z.string(), estimatedTokens: z.number(), }), async execute(input) { return { prompt: input.prompt }; }, }), { action: "prompt.completed", rules: (input) => [ tokenBudget({ key: "user123", // Replace with your authenticated user ID requested: Math.max(1, Math.ceil(input.estimatedTokens)), }), ], },);Then start or invoke the agent with a test prompt.
Requests appear in your Arcjet dashboard in real time.
In this example we use the Vercel AI SDK to create a simple AI chat endpoint with Next.js, and Arcjet to enforce per-user token budgets to prevent cost overruns. The same principles can be applied to any AI application, including those built with other frameworks.
We assume you already have a Next.js app set up.
Install the dependencies:
# Export your Arcjet API key from https://app.arcjet.comexport ARCJET_KEY="ajkey_..."
npm install @arcjet/next ai @ai-sdk/openaiCreate an AI chat endpoint:
import { openai } from "@ai-sdk/openai";import arcjet, { tokenBucket } from "@arcjet/next";import type { UIMessage } from "ai";import { convertToModelMessages, streamText } from "ai";
const aj = arcjet({ key: process.env.ARCJET_KEY!, // Get your site key from https://app.arcjet.com // Track budgets per user — replace "userId" with any stable identifier characteristics: ["userId"], rules: [ tokenBucket({ mode: "LIVE", // Blocks requests. Use "DRY_RUN" to log only refillRate: 2_000, // Refill 2,000 tokens per hour interval: "1h", capacity: 5_000, // Maximum 5,000 tokens in the bucket }), ],});
export async function POST(req: Request) { // Replace with your session/auth lookup to get a stable user ID const userId = "user-123"; const { messages }: { messages: UIMessage[] } = await req.json(); const modelMessages = await convertToModelMessages(messages);
// Estimate token cost: ~1 token per 4 characters of text (rough heuristic). // For accurate counts use https://www.npmjs.com/package/tiktoken const totalChars = modelMessages.reduce((sum, m) => { const content = typeof m.content === "string" ? m.content : JSON.stringify(m.content); return sum + content.length; }, 0); const estimate = Math.ceil(totalChars / 4);
// Deduct the estimated tokens from the user's budget const decision = await aj.protect(req, { userId, requested: estimate });
if (decision.isDenied()) { return new Response("AI usage limit exceeded", { status: 429 }); }
const result = await streamText({ model: openai("gpt-4o"), messages: modelMessages, });
return result.toUIMessageStreamResponse();}And hook it up to a chat UI:
"use client";
import { useChat } from "@ai-sdk/react";import { useState } from "react";
export default function Chat() { const [input, setInput] = useState(""); const [errorMessage, setErrorMessage] = useState<string | null>(null); const { messages, sendMessage } = useChat({ onError: async (e) => setErrorMessage(e.message), }); return ( <div className="flex flex-col w-full max-w-md py-24 mx-auto stretch"> {messages.map((message) => ( <div key={message.id} className="whitespace-pre-wrap"> {message.role === "user" ? "User: " : "AI: "} {message.parts.map((part, i) => { switch (part.type) { case "text": return <div key={`${message.id}-${i}`}>{part.text}</div>; } })} </div> ))}
{errorMessage && ( <div className="text-red-500 text-sm mb-4">{errorMessage}</div> )}
<form onSubmit={(e) => { e.preventDefault(); sendMessage({ text: input }); setInput(""); setErrorMessage(null); }} > <input className="fixed dark:bg-zinc-900 bottom-0 w-full max-w-md p-2 mb-8 border border-zinc-300 dark:border-zinc-800 rounded shadow-xl" value={input} placeholder="Say something..." onChange={(e) => setInput(e.currentTarget.value)} /> </form> </div> );}Then run the server:
npm run devRequests appear in your Arcjet dashboard in real time.
In this example we use LangChain to create a simple AI chat server with FastAPI, and Arcjet to enforce per-user token budgets to prevent cost overruns. The same principles can be applied to any AI application, including those built with other frameworks.
Set up the environment and install dependencies (uses uv, but you can also use pip to install the Arcjet Python SDK):
# Export your Arcjet API key from https://app.arcjet.comexport ARCJET_KEY="ajkey_..."export ARCJET_ENV=development
# Export your OpenAI API key (used by LangChain)export OPENAI_API_KEY="sk-..."
# Install dependenciesuv add arcjet fastapi uvicorn langchain langchain-openaiCreate the chat server:
import loggingimport mathimport os
from arcjet import Mode, arcjet, token_bucketfrom fastapi import FastAPI, Requestfrom fastapi.responses import JSONResponsefrom langchain_core.output_parsers import StrOutputParserfrom langchain_core.prompts import ChatPromptTemplatefrom langchain_openai import ChatOpenAIfrom pydantic import BaseModel
app = FastAPI()
logging.basicConfig(level=logging.INFO)logger = logging.getLogger(__name__)
arcjet_key = os.getenv("ARCJET_KEY")if not arcjet_key: raise RuntimeError("ARCJET_KEY is required. Get one at https://app.arcjet.com")
openai_api_key = os.getenv("OPENAI_API_KEY")if not openai_api_key: raise RuntimeError( "OPENAI_API_KEY is required. Get one at https://platform.openai.com" )
llm = ChatOpenAI(model="gpt-4o-mini", api_key=openai_api_key)
prompt = ChatPromptTemplate.from_messages( [ ("system", "You are a helpful assistant."), ("human", "{message}"), ])
chain = prompt | llm | StrOutputParser()
class ChatRequest(BaseModel): message: str
# Create a single Arcjet client at startup and reuse it across requestsaj = arcjet( key=arcjet_key, # Get your key from https://app.arcjet.com rules=[ # Token bucket rate limiting is best for AI budget control token_bucket( mode=Mode.LIVE, # Blocks requests. Use Mode.DRY_RUN to log only # Track budgets per user — replace "userId" with any stable # identifier. Removing this falls back to IP-based rate limiting. characteristics=["userId"], refill_rate=2_000, # Refill 2,000 tokens per interval interval=3_600, # Refill every hour (in seconds) capacity=5_000, # Maximum 5,000 tokens in the bucket ), ],)
@app.post("/chat")async def chat(request: Request, body: ChatRequest): # Replace with your session/auth lookup to get a stable user ID user_id = "user-123"
# Estimate token cost: ~1 token per 4 characters of text (rough heuristic). # For accurate counts use https://github.com/openai/tiktoken estimate = math.ceil(len(body.message) / 4)
# Deduct the estimated tokens from the user's budget decision = await aj.protect( request, requested=estimate, characteristics={"userId": user_id}, )
if decision.is_denied(): # The token_bucket rule is the only rule configured, so the only # possible denial reason is RATE_LIMIT (429). return JSONResponse({"error": "AI usage limit exceeded"}, status_code=429)
reply = await chain.ainvoke({"message": body.message})
return {"reply": reply}Then run the server:
uv run uvicorn main:app --reloadAnd send a message to the API endpoint:
curl -X POST http://localhost:8000/chat \ -H "Content-Type: application/json" \ -d '{"message": "What is the capital of France?"}'Requests appear in your Arcjet dashboard in real time.
In this example we use LangChain to create a simple AI chat server with Flask, and Arcjet to enforce per-user token budgets to prevent cost overruns. The same principles can be applied to any AI application, including those built with other frameworks.
Set up the environment and install dependencies (uses uv, but you can also use pip to install the Arcjet Python SDK):
# Export your Arcjet API key from https://app.arcjet.comexport ARCJET_KEY="ajkey_..."export ARCJET_ENV=development
# Export your OpenAI API key (used by LangChain)export OPENAI_API_KEY="sk-..."
# Install dependenciesuv add arcjet flask langchain langchain-openaiCreate the chat server:
import loggingimport mathimport os
from arcjet import Mode, arcjet_sync, token_bucketfrom flask import Flask, jsonify, requestfrom langchain_core.output_parsers import StrOutputParserfrom langchain_core.prompts import ChatPromptTemplatefrom langchain_openai import ChatOpenAI
app = Flask(__name__)
logging.basicConfig(level=logging.INFO)logger = logging.getLogger(__name__)
arcjet_key = os.getenv("ARCJET_KEY")if not arcjet_key: raise RuntimeError("ARCJET_KEY is required. Get one at https://app.arcjet.com")
openai_api_key = os.getenv("OPENAI_API_KEY")if not openai_api_key: raise RuntimeError( "OPENAI_API_KEY is required. Get one at https://platform.openai.com" )
llm = ChatOpenAI(model="gpt-4o-mini", api_key=openai_api_key)
prompt = ChatPromptTemplate.from_messages( [ ("system", "You are a helpful assistant."), ("human", "{message}"), ])
chain = prompt | llm | StrOutputParser()
# Create a single Arcjet client at startup and reuse it across requestsaj = arcjet_sync( key=arcjet_key, # Get your key from https://app.arcjet.com rules=[ # Token bucket rate limiting is best for AI budget control token_bucket( mode=Mode.LIVE, # Blocks requests. Use Mode.DRY_RUN to log only # Track budgets per user — replace "userId" with any stable # identifier. Removing this falls back to IP-based rate limiting. characteristics=["userId"], refill_rate=2_000, # Refill 2,000 tokens per interval interval=3_600, # Refill every hour (in seconds) capacity=5_000, # Maximum 5,000 tokens in the bucket ), ],)
@app.post("/chat")def chat(): # Replace with your session/auth lookup to get a stable user ID user_id = "user-123"
body = request.get_json() message = body.get("message", "") if body else ""
# Estimate token cost: ~1 token per 4 characters of text (rough heuristic). # For accurate counts use https://github.com/openai/tiktoken estimate = math.ceil(len(message) / 4)
# Deduct the estimated tokens from the user's budget decision = aj.protect( request, requested=estimate, characteristics={"userId": user_id}, )
if decision.is_denied(): # The token_bucket rule is the only rule configured, so the only # possible denial reason is RATE_LIMIT (429). return jsonify(error="AI usage limit exceeded"), 429
reply = chain.invoke({"message": message})
return jsonify(reply=reply)
if __name__ == "__main__": app.run(debug=True)Then run the server:
uv run python app.pyAnd send a message to the API endpoint:
curl -X POST http://localhost:5000/chat \ -H "Content-Type: application/json" \ -d '{"message": "What is the capital of France?"}'Requests appear in your Arcjet dashboard in real time.
Configure the rate limit
Section titled “Configure the rate limit”characteristics: ["userId"] - Tracks the bucket per user. Replace
"userId" with the characteristic that identifies a unique user in your
application, such as a session token, API key, or authenticated user ID. Pass the
value to aj.protect() as a named argument.
refillRate and interval - Set the sustained allowance. refillRate: 2_000, interval: "1h" gives each user 2,000 tokens per hour. Adjust to match your AI
provider’s pricing and your cost targets. These are hard coded in this example,
but you can also calculate them dynamically based on user subscription level or
other factors. Pass the calculated values to the rule.
capacity - The maximum tokens a user can accumulate. Setting capacity: 5_000 with refillRate: 2_000 lets users burst up to 5,000 tokens if they
haven’t used their allowance recently.
Token estimation
Section titled “Token estimation”The example uses a characters / 4 heuristic (~1 token per 4 characters for
common English text). This is a reasonable starting point – it avoids
introducing extra dependencies and works well enough for budget enforcement
where a small margin of error is acceptable.
For accurate counts, use a tokenizer:
- JavaScript / TypeScript:
tiktoken - Python:
tiktoken - Anthropic: provides a token counting API