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

This tutorial builds a small but production-ready chatbot service with FastAPI, SSE streaming, conversation memory, and model switching. The goal is not a toy demo. The goal is a backend you can put behind a real product surface and iterate on safely, without rewriting your integration every time you change models.

If you have already pointed one OpenAI-compatible SDK at TokenLab, this article picks up from there. If you have not done the base URL swap yet, read the migration guide first. If your main concern is request shaping and backoff under load, pair this guide with the AI API rate limiting guide.

Key Takeaways

What We Are Building

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 clear path from in-memory prototype to Redis or PostgreSQL.

That is enough to power a support bot, an internal assistant, or the first version of an embedded chat widget.

Install the Minimum Stack

pip install fastapi uvicorn openai pydantic redis

You can skip redis for the first pass, but wiring the import in now makes the later upgrade a non-event instead of a refactor.

Step 1: Start With a Small, Boring Chat Endpoint

The fastest way to get lost in chatbot work is starting with websockets, tool use, and agent orchestration before the basic request path is stable. Start with one small endpoint that proves your key, base URL, and model routing work.

from fastapi import FastAPI
from openai import OpenAI
from pydantic import BaseModel

app = FastAPI()

client = OpenAI(
    api_key="sk-tokenlab-xxx",
    base_url="https://api.tokenlab.sh/v1"
)

class ChatRequest(BaseModel):
    message: str
    model: str = "deepseek-v4-flash"
    conversation_id: str | None = None

@app.post("/chat")
async def chat(req: ChatRequest):
    response = client.chat.completions.create(
        model=req.model,
        messages=[{"role": "user", "content": req.message}]
    )
    return {"reply": response.choices[0].message.content}

Run one smoke test. If this fails, do not keep building on top of it.

Step 2: Add Streaming Because Users Feel Latency Before They Measure It

Most chatbot products feel slow not because the model is slow, but because the UI stays blank until the full response lands. SSE is enough for most chat products and has a lower operational burden than websockets.

from fastapi.responses import StreamingResponse

@app.post("/chat/stream")
async def chat_stream(req: ChatRequest):
    def generate():
        stream = client.chat.completions.create(
            model=req.model,
            messages=[{"role": "user", "content": req.message}],
            stream=True
        )
        for chunk in stream:
            delta = chunk.choices[0].delta
            if delta.content:
                yield f"data: {delta.content}\n\n"
        yield "data: [DONE]\n\n"

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

On the frontend, a plain fetch reader is still enough:

async function sendMessage(payload) {
  const response = await fetch('/chat/stream', {
    method: 'POST',
    headers: { 'Content-Type': 'application/json' },
    body: JSON.stringify(payload),
  });

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

  while (true) {
    const { done, value } = await reader.read();
    if (done) break;
    const chunk = decoder.decode(value, { stream: true });
    console.log(chunk);
  }
}

If your product already runs a browser client over standard HTTP, SSE keeps the architecture simpler than a websocket layer.

Step 3: Move Conversation State Out of the Request Body

The first chatbot demo usually keeps the full transcript in the browser and resends it on every turn. That works for prototypes. It gets messy fast once you need retries, resumable sessions, or server-side tooling.

An in-memory store is fine to start:

from collections import defaultdict
import uuid

conversations: dict[str, list] = defaultdict(list)
SYSTEM_PROMPT = "You are a helpful assistant. Be concise and direct."

def build_messages(conv_id: str, user_msg: str) -> list:
    messages = [{"role": "system", "content": SYSTEM_PROMPT}]
    history = conversations[conv_id][-20:]
    messages.extend(history)
    messages.append({"role": "user", "content": user_msg})
    conversations[conv_id].append({"role": "user", "content": user_msg})
    return messages

The upgrade path to Redis is mostly storage plumbing, not logic changes:

import json
import redis

redis_client = redis.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))

Reach for Redis when conversations need TTL, resumability, or multi-instance deployment. Reach for PostgreSQL when the transcript itself is product data you need to query, audit, or report on later.

Step 4: Treat Errors as Product Behavior, Not Just Exceptions

If your chatbot is customer-facing, the failure path matters as much as the happy path. A user does not care whether the failure came from rate limiting, an exhausted balance, or a model outage upstream. They care whether the UI freezes.

from openai import APIConnectionError, APIError, RateLimitError

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

    def generate():
        full_response = []
        try:
            stream = client.chat.completions.create(
                model=req.model,
                messages=messages,
                stream=True
            )
            for chunk in stream:
                delta = chunk.choices[0].delta
                if delta.content:
                    full_response.append(delta.content)
                    yield f"data: {delta.content}\n\n"
        except RateLimitError:
            yield "data: [error: rate limited, please retry shortly]\n\n"
        except APIConnectionError:
            yield "data: [error: connection issue, please retry]\n\n"
        except APIError:
            yield "data: [error: something went wrong on our end]\n\n"
        finally:
            if full_response:
                conversations[conv_id].append(
                    {"role": "assistant", "content": "".join(full_response)}
                )
        yield "data: [DONE]\n\n"

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

Step 5: Lock Down Which Models the Frontend Can Request

Never let the frontend pass an arbitrary model string straight to the API. A free text field invites requests for retired models, typos, or models you never intended to expose. Route through a backend allowlist instead.

ALLOWED_MODELS = {
    "default": "deepseek-v4-flash",
    "flagship": "gpt-5.5",
    "balanced": "claude-sonnet-5",
}

class ChatRequest(BaseModel):
    message: str
    tier: str = "default"
    conversation_id: str | None = None

@app.post("/chat/stream")
async def chat_stream(req: ChatRequest):
    model = ALLOWED_MODELS.get(req.tier, ALLOWED_MODELS["default"])
    # rest of the streaming logic uses `model` instead of a raw client-supplied string

This gives you a single place to swap models when a provider deprecates one, without touching frontend code or shipping a client update.

Step 6: Handle the Rest of Production, Not Just the Happy Path

A chatbot backend is considered production-ready when the surrounding edges are handled, not when the core chat call gets clever.

The checklist is short:

• add request IDs so you can connect frontend failures to backend logs
• cap per-user concurrency and request size
• trim long histories before they explode your token budget
• log model, latency, input size, and finish reason
• separate user-visible error messages from internal error detail
• test one alternate model so you know fallback works before the first real outage

History trimming can stay simple:

def trim_history(messages: list, max_tokens: int = 8000) -> list:
    system = messages[0]
    history = messages[1:]
    total_chars = len(system["content"])
    trimmed = []

    for msg in reversed(history):
        msg_chars = len(msg["content"])
        if total_chars + msg_chars > max_tokens * 4:
            break
        trimmed.insert(0, msg)
        total_chars += msg_chars

    return [system] + trimmed

The point is not token-perfect accounting. The point is stopping obvious context blowups before they hit your bill or your latency.

From Demo to Product

Once this backend is stable, the next upgrade is rarely "more AI." It's usually boring infrastructure:

• auth so one user cannot read another user's conversation
• persistence so sessions survive deploys
• rate limiting so one noisy user cannot burn your quota
• billing or usage attribution if the chatbot is customer-facing
• background summarization if conversations need long-term memory

A unified gateway helps with most of this. Once the base URL migration is behind you, model changes stop being a platform rewrite and become a config edit.

Smoke Test

uvicorn main:app --reload --port 8000

curl -N -X POST http://localhost:8000/chat/stream \
  -H "Content-Type: application/json" \
  -d '{"message": "Hello!", "model": "deepseek-v4-flash"}'

If you can stream one turn, preserve one conversation, and return a clean error on a forced failure, you have the right foundation.

Cost Estimate

Create an API key at TokenLab, point your OpenAI SDK at https://api.tokenlab.sh/v1, and you can ship the first production version of your chatbot without managing separate accounts across providers.

Model	Typical Tier	Notes
DeepSeek V4 Flash	Fast / default	Good default for high-volume, low-latency turns
GPT-5.5	Flagship	Use for turns that need stronger reasoning
Claude Sonnet 5	Balanced	Strong pick for coding and review-style replies
Gemini 3.5 Flash	Budget / fast alt	Low-cost, fast alternative for high-volume routing

Exact per-token pricing changes frequently across providers and is not reproduced here as fixed numbers. Check current rates on the model directory (observed 2026-07-07) before you budget against them. In practice, routing most conversations to a fast/default tier like DeepSeek V4 Flash or Gemini 3.5 Flash and reserving GPT-5.5 or Claude Sonnet 5 for turns that need it keeps most applications on a low monthly bill, but confirm the actual per-million-token rates for your account before committing to a budget.

FAQ

Do I need websockets to build an AI chatbot? No. SSE streaming, shown in Step 2, covers the vast majority of chat products. Websockets add real value when you need bidirectional push outside of request/response, like live collaboration or server-initiated events. For a standard chat UI, SSE is simpler to deploy, debug, and scale.

How do I know which model to default to? Start with a fast, low-cost model such as DeepSeek V4 Flash or Gemini 3.5 Flash for the default tier, and add a balanced or reasoning tier on Claude Sonnet 5 or GPT-5.5 behind the allowlist shown in Step 5. Check the model directory (observed 2026-07-07) for current options, since new models ship and older ones get deprecated on a schedule outside your control.

What breaks first when a chatbot goes from demo to real traffic? Almost always the error path, not the happy path. Unbounded retries, missing per-user concurrency caps, and unbounded conversation history are the three most common causes of a chatbot backend falling over under real load. Steps 4 and 6 above address all three directly.

---

Get Started with your API key: tokenlab.sh provides 300+ models through one endpoint. $1 free credit to start building.