Build an AI Chatbot with One API Key: From Zero to Production in 30 Minutes

You can build an AI chatbot that streams one reply in an afternoon. Keeping it alive after the first 429 is the real work. We build a small FastAPI service with SSE streaming, server-side conversation memory, and a model allowlist, all behind one TokenLab key. The goal is a backend you can put behind a real product surface and change models in without rewriting the integration. If you have not done the base URL swap yet, read the migration guide first. If your main concern is backoff under load, pair this guide with the AI API rate limiting guide.

Key Takeaways

Build an AI Chatbot: Setup and the Core Service

The finished service has six moving parts:

1. A synchronous /chat endpoint for smoke tests.
2. A streaming /chat/stream endpoint for the real UI.
3. Conversation state keyed by conversation_id.
4. A model allowlist, so the frontend cannot request arbitrary IDs.
5. Error handling that does not collapse on the first 429.
6. A path from in-memory prototype to Redis or PostgreSQL.

That is enough for a support bot, an internal assistant, or the first version of an embedded chat widget. Each code block below appends to one file, main.py, in the order shown.

Step 1: Set the key and install the stack

Create a key under API Keys in your workspace. The full value is shown only once, and the prefix in the key list cannot authenticate requests. All TokenLab keys start with sk-. Check your balance in Billing as well, because the first request is billed at the model's current price (Quickstart, observed 2026-10-03).

Store the key in an environment variable, never in source. The prompt below keeps it out of the command line and shell history, and it applies only to the current terminal and its child processes. Use a secret manager for deployed apps. The Quickstart also has a PowerShell version.

printf 'TokenLab API key: '
read -r -s TOKENLAB_API_KEY
printf '\n'
export TOKENLAB_API_KEY

pip install fastapi uvicorn openai pydantic

You do not need a TokenLab-specific SDK. The official OpenAI SDK works once you change the key and base URL (SDKs guide, observed 2026-10-03).

Step 2: Define the schema, allowlist, and memory

We start with the shared pieces. The request schema uses tier, not a raw model string, and every curl example later in this guide sends the same fields.

The first chatbot demo usually keeps the transcript in the browser and resends it each turn. For example, that works for a prototype. It gets messy once you need retries, resumable sessions, or server-side tooling. So the server owns the history from the start.

import json
import logging
import os
import time
import uuid
from collections import defaultdict

from fastapi import FastAPI, HTTPException
from fastapi.responses import StreamingResponse
from openai import APIConnectionError, APIStatusError, OpenAI, RateLimitError
from pydantic import BaseModel, Field

log = logging.getLogger("chatbot")
app = FastAPI()

client = OpenAI(
    api_key=os.environ["TOKENLAB_API_KEY"],
    base_url="https://api.tokenlab.sh/v1",
    timeout=30.0,
    max_retries=0,
)

ALLOWED_MODELS = {
    "default": "deepseek-v4-flash",
    "flagship": "gpt-5.5",
    "balanced": "claude-sonnet-5",
    "budget": "gemini-3.5-flash",
}
SYSTEM_PROMPT = "You are a helpful assistant. Be concise and direct."
MAX_HISTORY = 20          # messages kept per conversation (your choice)
MAX_REPLY_TOKENS = 1000   # output cap per turn (your choice)

class ChatRequest(BaseModel):
    message: str = Field(min_length=1, max_length=4000)
    tier: str = "default"
    conversation_id: str | None = None

def resolve_model(tier: str) -> str:
    model = ALLOWED_MODELS.get(tier)
    if model is None:
        raise HTTPException(status_code=400, detail=f"unknown tier: {tier}")
    return model

conversations: dict[str, list[dict]] = defaultdict(list)

def build_messages(conv_id: str, user_msg: str) -> list[dict]:
    history = conversations[conv_id][-MAX_HISTORY:]
    return [
        {"role": "system", "content": SYSTEM_PROMPT},
        *history,
        {"role": "user", "content": user_msg},
    ]

def remember(conv_id: str, user_msg: str, reply: str) -> None:
    conversations[conv_id].append({"role": "user", "content": user_msg})
    conversations[conv_id].append({"role": "assistant", "content": reply})

The client sets max_retries=0 and a 30-second timeout, matching the Quickstart. We make retry decisions ourselves, because a blind SDK retry on a streamed turn can repeat a charge.

Step 3: Map errors and add the sync endpoint

We keep one small function that turns any SDK exception into a status code and a user-safe message. The detail goes to your logs, not to the user.

def describe_error(exc: Exception) -> tuple[int, str]:
    if isinstance(exc, RateLimitError):  # check first: it subclasses APIStatusError
        return 429, "The assistant is busy. Please retry shortly."
    if isinstance(exc, APIConnectionError):
        return 502, "Connection problem. Please retry."
    if isinstance(exc, APIStatusError):
        log.error("upstream status=%s code=%s", exc.status_code, exc.code)
        if exc.status_code in (401, 402, 403):  # key, balance, or permission: operator problem
            return 503, "The assistant is unavailable right now."
        if exc.status_code in (400, 404):  # bad field or model ID: our bug
            return 500, "Something went wrong on our side."
        return 502, "The assistant is unavailable right now."
    return 500, "Something went wrong."

def retry_delay(exc: RateLimitError) -> float:
    try:
        return min(float(exc.response.headers.get("retry-after", "2")), 10.0)
    except ValueError:
        return 2.0

def create_with_one_retry(**kwargs):
    try:
        return client.chat.completions.create(**kwargs)
    except RateLimitError as exc:
        time.sleep(retry_delay(exc))
        return client.chat.completions.create(**kwargs)

@app.post("/chat")
def chat(req: ChatRequest):
    model = resolve_model(req.tier)
    conv_id = req.conversation_id or str(uuid.uuid4())
    messages = build_messages(conv_id, req.message)
    try:
        response = create_with_one_retry(
            model=model, messages=messages, max_tokens=MAX_REPLY_TOKENS
        )
    except (APIConnectionError, APIStatusError) as exc:
        status, message = describe_error(exc)
        raise HTTPException(status_code=status, detail=message)
    reply = response.choices[0].message.content or ""
    log.info("model=%s usage=%s", model, response.usage)
    remember(conv_id, req.message, reply)
    return {"conversation_id": conv_id, "reply": reply}

Run one request before building anything else. If this fails, do not keep building on top of it.

Stream Replies and Keep the AI Chatbot Responsive

Most chatbot products feel slow not because the model is slow, but because the UI stays blank until the full response lands. We add SSE because it covers most chat products and carries less operational weight than websockets. If your product already runs a browser client over plain HTTP, SSE keeps the architecture simpler.

Step 4: Add the streaming endpoint

The streaming docs (Streaming guide, observed 2026-10-03) keep Chat Completions streaming valid for frameworks that already expect SSE chunks from /v1/chat/completions. We stay on that path because all four allowlisted models accept it. We encode each event as JSON, so newlines inside a reply cannot break the framing.

def sse(payload: dict) -> str:
    return f"data: {json.dumps(payload)}\n\n"

@app.post("/chat/stream")
def chat_stream(req: ChatRequest):
    model = resolve_model(req.tier)
    conv_id = req.conversation_id or str(uuid.uuid4())
    messages = build_messages(conv_id, req.message)

    def generate():
        parts: list[str] = []
        finish_reason = None
        try:
            with client.chat.completions.create(
                model=model,
                messages=messages,
                max_tokens=MAX_REPLY_TOKENS,
                stream=True,
                stream_options={"include_usage": True},
            ) as stream:
                for chunk in stream:
                    if chunk.usage:
                        log.info("model=%s usage=%s", model, chunk.usage)
                    if not chunk.choices:
                        continue
                    choice = chunk.choices[0]
                    if choice.delta.content:
                        parts.append(choice.delta.content)
                        yield sse({"delta": choice.delta.content})
                    if choice.finish_reason:
                        finish_reason = choice.finish_reason
        except (APIConnectionError, APIStatusError) as exc:
            _, message = describe_error(exc)
            yield sse({"error": message})
            return
        if finish_reason != "stop":
            yield sse({"error": f"Reply incomplete ({finish_reason})."})
            return
        remember(conv_id, req.message, "".join(parts))
        yield sse({"done": True, "conversation_id": conv_id})

    return StreamingResponse(generate(), media_type="text/event-stream")

Two details matter here. First, include_usage: true adds token usage to the stream chunks, and the final usage chunk has no choices, so the loop skips it. Second, we save the turn only when finish_reason is stop. A length finish means the token limit cut the reply off, and the docs list both as normal end conditions.

Because errors surface inside the generator, the HTTP status is already 200 by the time a failure happens. The client must read the error event. The with block closes the upstream stream when the generator exits. The docs advise cancelling upstream when the user stops or disconnects, so test that your ASGI server actually stops the generator on disconnect.

Step 5: Read the stream in the browser

The browser should call your own authenticated backend and never TokenLab directly. Parse by event boundary, because one network chunk is not necessarily one complete SSE event.

async function sendMessage(payload, onDelta) {
  const response = await fetch('/chat/stream', {
    method: 'POST',
    headers: { 'Content-Type': 'application/json' },
    body: JSON.stringify(payload), // { message, tier, conversation_id }
  });
  if (!response.ok || !response.body) throw new Error(`HTTP ${response.status}`);

  const reader = response.body.getReader();
  const decoder = new TextDecoder();
  let buffer = '';

  while (true) {
    const { done, value } = await reader.read();
    if (done) break;
    buffer += decoder.decode(value, { stream: true });
    const events = buffer.split('\n\n');
    buffer = events.pop();
    for (const raw of events) {
      if (!raw.startsWith('data: ')) continue;
      const event = JSON.parse(raw.slice(6));
      if (event.error) throw new Error(event.error);
      if (event.delta) onDelta(event.delta);
    }
  }
}

Upgrade memory to Redis when you need it

The in-memory dictionary is fine for one process. Reach for Redis when conversations need a TTL, resumability, or multiple instances. Reach for PostgreSQL when the transcript is product data you must query, audit, or report on. The Redis change is storage plumbing, not logic. Run pip install redis, then have build_messages and remember read and write through these helpers.

import redis

redis_client = redis.Redis(
    host=os.environ.get("REDIS_HOST", "127.0.0.1"), port=6379, decode_responses=True
)

def load_history(conv_id: str) -> list:
    raw = redis_client.get(f"chat:{conv_id}")
    return json.loads(raw) if raw else []

def save_history(conv_id: str, history: list) -> None:
    redis_client.setex(f"chat:{conv_id}", 60 * 60 * 24, json.dumps(history))

Handle Errors When You Build an AI Chatbot for Production

A customer-facing user does not care whether a failure came from rate limiting, an empty balance, or an upstream outage. They care whether the UI freezes. Imagine a support widget whose stream dies after forty words. The TokenLab docs say a stream interrupted after output arrived is not a complete response, and repeating it may generate different output and a second charge (Handle API errors, observed 2026-10-03). That is why Step 4 never saves an unfinished turn.

Step 6: Apply the retry rules

Handle errors by HTTP status and code, not by the human-readable message, which can change without notice. The table combines the docs with what our service does.

Status	Meaning (per docs)	What this service does
400	Field, model ID, or input is invalid	Return 500, log it, never retry
401	Key missing, invalid, expired, or revoked	Return 503, alert the operator
402	Balance or API-key limit too low (insufficient_balance, quota_exceeded)	Return 503, alert the operator
403	Key cannot use this model	Return 503, check the key's model list
404	Resource or model not found	Return 500, check the ID against /v1/models
429	Request limit reached	Wait for Retry-After, retry once (sync only)
500–504	Service unavailable or network failure	Retry only when retryable is true; respect retry_after

Two cautions from the docs. A 503 all_channels_failed does not always mean a temporary outage. If retryable is false, do not repeat the same request, and check GET /v1/models before picking another model. Also, Chat Completions and Responses use an OpenAI-style error object, but Anthropic Messages and Gemini keep their own formats. Our service calls Chat Completions on every allowlisted model, so one parser works. If you move a tier to a native format, you need a second parser.

Save the Request ID from the response headers with the endpoint, model, time, and your own user or job ID. The docs do not name the header on the pages we used, so confirm it against the troubleshooting path before you wire up your logs.

The rest of the production checklist is short:

• Add your own request IDs so you can connect frontend failures to backend logs.
• Cap per-user concurrency and request size. The max_length=4000 on message is a starting point you should tune.
• Cap history by message count, as MAX_HISTORY does. The billing docs say tokenization differs by model, language, and media input, so a character-count formula is not a reliable billing estimate.
• Log model, latency, usage, and finish reason on every turn.
• Keep user-visible messages separate from internal error detail.
• Send one test turn to an alternate tier, so you know fallback works before the first real outage.

Choose Models and Estimate Cost for Your AI Chatbot

Never let the frontend pass an arbitrary model string to the API. A free-text field invites typos, retired models, and models you never meant to expose. The ALLOWED_MODELS map gives you one place to swap a model when a provider deprecates one, with no client update. Unknown model IDs return model_not_found, and the docs point you to /v1/models for a current ID. Check the model directory before you lock the allowlist.

Current per-1M-token prices

These are TokenLab prices in USD from each model's live API entry, observed 2026-10-03. The pricing was last updated 2026-10-02T16:53:30.068Z. All four models showed lifecycle stage active.

Tier	Model ID	Input / 1M	Output / 1M	Cache read / 1M	Source
default	deepseek-v4-flash (off-peak)	$0.15	$0.60	$0.003	model API
default	deepseek-v4-flash (peak)	$0.30	$1.20	$0.006	model API
flagship	gpt-5.5	$1.50	$9.00	$0.15	model API
balanced	claude-sonnet-5	$0.90	$4.50	$0.09	model API
budget	gemini-3.5-flash	$0.75	$4.50	$0.075	model API

For deepseek-v4-flash, the entry lists peak windows on weekdays, excluding China public holidays. Off-peak includes weekends and those holidays. claude-sonnet-5 also lists a cache write price of $1.125 per 1M tokens.

One worked request

Assume one turn with 2,000 input tokens and 500 output tokens. That size is our assumption, not a measurement. For gpt-5.5:

• Input: 2,000 × $1.50 / 1,000,000 = $0.0030
• Output: 500 × $9.00 / 1,000,000 = $0.0045
• Total: $0.0075 per turn (estimate, no cache hits)

Using the same arithmetic, we get these estimates: deepseek-v4-flash costs $0.0006 off-peak and $0.0012 at peak, claude-sonnet-5 costs $0.00405, and gemini-3.5-flash costs $0.00375. Over 1,000 such turns, that is roughly $7.50 on gpt-5.5 against $0.60 to $1.20 on deepseek-v4-flash. Real turns grow as history accumulates, so read usage from the response or the final stream chunk rather than trusting this sizing.

In practice, we route most conversations to the default tier and reserve gpt-5.5 or claude-sonnet-5 for turns that need them. The billing docs warn that the lowest price per token is not always the lowest cost per completed task. Run the same prompts through two tiers and compare the final cost in Usage.

Two checks before you ship. The billing docs say not to hard-code a copied price table, so read GET /v1/models/:model/pricing if your app displays costs (Billing and pricing, observed 2026-10-03). Also, the deepseek-v4-flash entry lists json-mode, prompt-cache, and tool-use but not vision. If your chatbot accepts images, route those turns to a tier whose entry lists vision.

Smoke Test and Ship Your AI Chatbot

Step 7: Run it end to end

Start the server in a terminal where TOKENLAB_API_KEY is exported. The curl payloads use the same message, tier, and conversation_id fields as ChatRequest.

uvicorn main:app --reload --port 8000

# 1. Sync endpoint: expect a short reply and a conversation_id
curl -X POST http://localhost:8000/chat \
  -H "Content-Type: application/json" \
  -d '{"message": "Reply only with OK.", "tier": "default"}'

# 2. Streaming: expect data: {"delta": ...} lines, then {"done": true, ...}
curl -N -X POST http://localhost:8000/chat/stream \
  -H "Content-Type: application/json" \
  -d '{"message": "Hello!", "tier": "default"}'

# 3. Memory: reuse a conversation_id returned above
curl -X POST http://localhost:8000/chat \
  -H "Content-Type: application/json" \
  -d '{"message": "What did I just say?", "tier": "default", "conversation_id": "PASTE_ID_FROM_STEP_1"}'

# 4. Forced failure: expect HTTP 400 and "unknown tier: nope"
curl -i -X POST http://localhost:8000/chat \
  -H "Content-Type: application/json" \
  -d '{"message": "Hello!", "tier": "nope"}'

To force an upstream failure, restart the server with a wrong TOKENLAB_API_KEY. Your client should get a clean 503 message, and your logs should show the upstream 401 with invalid_api_key. Then open Requests to confirm the model, status, and usage of the successful calls. If you can stream one turn, preserve one conversation, and return a clean error on a forced failure, you have the right foundation.

After that, the next upgrades are boring infrastructure rather than more AI:

• Auth, so one user cannot read another user's conversation.
• Persistence, so sessions survive deploys.
• Per-user rate limiting, so one noisy user cannot burn your quota.
• Usage attribution, if the chatbot is customer-facing.
• Background summarization, if conversations need long-term memory.

Once the base URL migration is behind you, a model change is a config edit to ALLOWED_MODELS.

FAQ

Do I need websockets to build an AI chatbot?

No. SSE, shown in Step 4, covers most chat products. Websockets add value when you need bidirectional push outside request and response, such as live collaboration or server-initiated events. For a standard chat UI, SSE is simpler to deploy and debug.

Which model should my chatbot use by default?

Start with a low-priced tier such as deepseek-v4-flash and add claude-sonnet-5 or gpt-5.5 behind the allowlist for harder turns. The prices above, observed 2026-10-03, show the gap. Then test your own prompts on two tiers and compare the final cost in Usage. Check the model directory before you commit, since lineups change.

Why does my chatbot return 402 or 429?

A 402 means the account balance or the key's own usage limit is too low. The codes are insufficient_balance and quota_exceeded, and repeating the request will not help. A 429 means you hit a request limit, so wait for Retry-After and then retry. Our sync endpoint retries once and the streaming endpoint reports the error to the client.

Can the browser call TokenLab directly?

Not safely. The Authentication docs say never to expose API keys in client-side code. Keep the key in the server environment and have the browser call your own authenticated backend, as the frontend sample does.

---

Create an API key at tokenlab.sh, add funds, and run the smoke test above.