How to Build an AI Chatbot with Streaming Responses: A Complete Developer Guide
Real-time streaming is no longer optional for AI chatbots. Users expect to see text appear word by word, just like they do in ChatGPT and Claude. In this tutorial, you will learn how to build a production-ready AI chatbot with streaming responses using any major LLM API, with full code examples in Python and JavaScript.
Why Streaming Matters
When you send a prompt to an AI model, generating the full response can take 5-30 seconds depending on the length. Without streaming, your user stares at a loading spinner the entire time. With streaming, text starts appearing almost immediately, creating a dramatically better user experience.
The benefits of streaming include:
- Perceived performance — Users see results instantly instead of waiting
- Cancelability — Users can stop generation mid-stream if the response goes off track
- Progressive rendering — You can render markdown and code blocks as they arrive
- Lower timeout risk — Chunks arrive continuously, avoiding HTTP timeout issues
Prerequisites
Before we start, you will need:
- An API key from Qubax AI (works with GPT-5, Claude, Gemini, and 100+ models)
- Python 3.11+ or Node.js 20+
- Basic familiarity with async/await patterns
Part 1: Streaming Chatbot in Python (FastAPI)
Let us build a FastAPI backend that streams AI responses to a web frontend using Server-Sent Events (SSE).
Step 1: Install Dependencies
pip install fastapi uvicorn httpx sse-starlette python-dotenvStep 2: Set Up Your Environment
Create a .env file:
QUBAX_API_KEY=your_api_key_here
QUBAX_BASE_URL=https://api.qubax.ai/v1Step 3: Create the Streaming Backend
# main.py
import os
import json
import httpx
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
from sse_starlette.sse import EventSourceResponse
from pydantic import BaseModel
from dotenv import load_dotenv
load_dotenv()
app = FastAPI(title="AI Chatbot API")
app.add_middleware(
CORSMiddleware,
allow_origins=["*"],
allow_methods=["*"],
allow_headers=["*"],
)
API_KEY = os.getenv("QUBAX_API_KEY")
BASE_URL = os.getenv("QUBAX_BASE_URL", "https://api.qubax.ai/v1")
class ChatMessage(BaseModel):
role: str
content: str
class ChatRequest(BaseModel):
messages: list[ChatMessage]
model: str = "gpt-5"
temperature: float = 0.7
max_tokens: int = 2000
@app.post("/chat/stream")
async def chat_stream(request: ChatRequest):
# Stream AI responses using Server-Sent Events
async def event_generator():
headers = {
"Authorization": f"Bearer {API_KEY}",
"Content-Type": "application/json",
}
payload = {
"model": request.model,
"messages": [m.model_dump() for m in request.messages],
"stream": True,
"temperature": request.temperature,
"max_tokens": request.max_tokens,
}
async with httpx.AsyncClient(timeout=120.0) as client:
async with client.stream(
"POST", f"{BASE_URL}/chat/completions",
headers=headers, json=payload
) as response:
async for line in response.aiter_lines():
if line.startswith("data: "):
data = line[6:]
if data == "[DONE]":
yield {"event": "done", "data": ""}
break
try:
chunk = json.loads(data)
content = chunk["choices"][0]["delta"].get("content", "")
if content:
yield {"event": "token", "data": content}
except json.JSONDecodeError:
continue
return EventSourceResponse(event_generator())
@app.get("/models")
async def list_models():
headers = {"Authorization": f"Bearer {API_KEY}"}
async with httpx.AsyncClient() as client:
response = await client.get(f"{BASE_URL}/models", headers=headers)
return response.json()Step 4: Run the Server
uvicorn main:app --reload --port 8000Part 2: Streaming Chatbot in JavaScript (Node.js)
Prefer JavaScript? Here is the same streaming chatbot using Express and the native Fetch API:
import express from "express";
import cors from "cors";
import dotenv from "dotenv";
dotenv.config();
const app = express();
app.use(cors());
app.use(express.json());
const API_KEY = process.env.QUBAX_API_KEY;
const BASE_URL = process.env.QUBAX_BASE_URL || "https://api.qubax.ai/v1";
app.post("/chat/stream", async (req, res) => {
const { messages, model = "gpt-5" } = req.body;
res.setHeader("Content-Type", "text/event-stream");
res.setHeader("Cache-Control", "no-cache");
res.setHeader("Connection", "keep-alive");
try {
const response = await fetch(BASE_URL + "/chat/completions", {
method: "POST",
headers: {
Authorization: "Bearer " + API_KEY,
"Content-Type": "application/json"
},
body: JSON.stringify({ model, messages, stream: true }),
});
const reader = response.body.getReader();
const decoder = new TextDecoder();
while (true) {
const { done, value } = await reader.read();
if (done) break;
const text = decoder.decode(value);
const lines = text.split("\n").filter(l => l.startsWith("data: "));
for (const line of lines) {
const data = line.slice(6);
if (data === "[DONE]") {
res.write("event: done\n\n");
res.end();
return;
}
try {
const chunk = JSON.parse(data);
const delta = chunk.choices[0] && chunk.choices[0].delta;
const content = (delta && delta.content) || "";
if (content) {
res.write("event: token\ndata: " + JSON.stringify(content) + "\n\n");
}
} catch (e) {}
}
}
res.end();
} catch (error) {
res.write("event: error\ndata: " + JSON.stringify(error.message) + "\n\n");
res.end();
}
});
app.listen(3001, () => console.log("Chatbot server running"));Part 3: Building the Frontend
Here is a minimal HTML and JS frontend that connects to your streaming backend:
<!DOCTYPE html>
<html>
<head>
<title>AI Chatbot</title>
<style>
body { font-family: system-ui; max-width: 800px; margin: 40px auto; padding: 20px; }
#chat { height: 400px; overflow-y: auto; border: 1px solid #ddd; padding: 20px; border-radius: 8px; }
.message { margin-bottom: 12px; padding: 8px 12px; border-radius: 8px; }
.user { background: #007bff; color: white; text-align: right; }
.assistant { background: #f1f1f1; }
.input-area { display: flex; gap: 10px; margin-top: 20px; }
#input { flex: 1; padding: 10px; border: 1px solid #ddd; border-radius: 8px; }
button { padding: 10px 20px; background: #007bff; color: white; border: none; border-radius: 8px; cursor: pointer; }
</style>
</head>
<body>
<h1>AI Chatbot</h1>
<div id="chat"></div>
<div class="input-area">
<input id="input" placeholder="Type your message..." />
<button onclick="sendMessage()">Send</button>
<button onclick="stopGeneration()" id="stopBtn" style="display:none;">Stop</button>
</div>
<script>
let messages = [];
let currentController = null;
async function sendMessage() {
const input = document.getElementById("input");
const text = input.value.trim();
if (!text) return;
messages.push({ role: "user", content: text });
input.value = "";
displayMessage("user", text);
var assistantDiv = displayMessage("assistant", "");
currentController = new AbortController();
document.getElementById("stopBtn").style.display = "inline";
try {
var response = await fetch("/chat/stream", {
method: "POST",
headers: { "Content-Type": "application/json" },
body: JSON.stringify({ messages: messages, model: "gpt-5" }),
signal: currentController.signal,
});
var reader = response.body.getReader();
var decoder = new TextDecoder();
var assistantText = "";
var buffer = "";
while (true) {
var result = await reader.read();
if (result.done) break;
buffer += decoder.decode(result.value);
var lines = buffer.split("\n");
buffer = lines.pop();
for (var i = 0; i < lines.length; i++) {
var line = lines[i];
if (line.startsWith("data:")) {
var data = line.slice(5).trim();
if (data && data !== "[DONE]") {
try {
var token = JSON.parse(data);
assistantText += token;
assistantDiv.textContent = assistantText;
} catch (e) {}
}
}
}
}
messages.push({ role: "assistant", content: assistantText });
} catch (e) {
if (e.name !== "AbortError") assistantDiv.textContent = "Error: " + e.message;
} finally {
document.getElementById("stopBtn").style.display = "none";
currentController = null;
}
}
function displayMessage(role, content) {
var chat = document.getElementById("chat");
var div = document.createElement("div");
div.className = "message " + role;
div.textContent = content;
chat.appendChild(div);
chat.scrollTop = chat.scrollHeight;
return div;
}
function stopGeneration() {
if (currentController) {
currentController.abort();
document.getElementById("stopBtn").style.display = "none";
}
}
document.getElementById("input").addEventListener("keypress", function(e) {
if (e.key === "Enter") sendMessage();
});
</script>
</body>
</html>Part 4: Advanced Features
Conversation History Management
class ConversationManager:
def __init__(self, max_messages=20):
self.max_messages = max_messages
self.conversations = {}
def add_message(self, session_id, role, content):
if session_id not in self.conversations:
self.conversations[session_id] = []
self.conversations[session_id].append({"role": role, "content": content})
if len(self.conversations[session_id]) > self.max_messages:
self.conversations[session_id] = (
self.conversations[session_id][:1] +
self.conversations[session_id][-self.max_messages:]
)
def get_messages(self, session_id):
return self.conversations.get(session_id, [])Error Handling and Retries
import asyncio
async def robust_stream_completion(messages, model="gpt-5", max_retries=3, api_key=None):
for attempt in range(max_retries):
try:
async with httpx.AsyncClient(timeout=120.0) as client:
async with client.stream(
"POST", "https://api.qubax.ai/v1/chat/completions",
headers={"Authorization": f"Bearer {api_key}"},
json={"model": model, "messages": messages, "stream": True}
) as response:
if response.status_code == 429:
retry_after = int(response.headers.get("retry-after", 5))
await asyncio.sleep(retry_after)
continue
response.raise_for_status()
async for line in response.aiter_lines():
if line.startswith("data: ") and line[6:] != "[DONE]":
chunk = json.loads(line[6:])
token = chunk["choices"][0]["delta"].get("content", "")
if token:
yield token
return
except (httpx.ConnectError, httpx.ReadTimeout):
if attempt < max_retries - 1:
await asyncio.sleep(2 ** attempt)
else:
raiseCost Optimization Tips
Streaming does not change the total cost — you still pay for the same number of tokens. But you can optimize costs by:
- Use cheaper models for simple tasks — Models like GPT-5 Mini or DeepSeek cost 90% less than premium models
- Set max_tokens — Prevent runaway responses that waste tokens
- Cache common responses — Use Redis to cache identical queries
- Compress context — Summarize old conversation history instead of sending it all
Compare model pricing at Qubax AI models to find the most cost-effective option for your use case.
Deployment Checklist
Before deploying your streaming chatbot to production:
- [ ] Enable CORS only for your actual frontend domain (not
*) - [ ] Add rate limiting per user (e.g., 20 messages per minute)
- [ ] Implement user authentication
- [ ] Add logging and monitoring
- [ ] Set up health checks
- [ ] Configure proper error responses
- [ ] Test with concurrent users
- [ ] Use environment variables for all secrets
- [ ] Add a system prompt to control AI behavior
For the full API reference, see the Qubax AI documentation.
Build your AI chatbot today with [Qubax AI](https://qubax.ai/models). Get a single API key for 100+ AI models, streaming support out of the box, and pricing that saves you up to 90% compared to going direct. Sign up free.
FAQ
### What is streaming in AI chatbots?
Streaming means the AI sends its response in small chunks (tokens) as they are generated, rather than waiting for the entire response to complete. This creates a much faster perceived response time.
### Do I need a special API for streaming?
Most modern AI APIs support streaming via Server-Sent Events (SSE). The Qubax AI API supports streaming for all models — just set stream: true in your request.
### Can I stream responses in any programming language?
Yes. Any language that supports HTTP requests can consume SSE streams. This tutorial shows examples in Python and JavaScript, but the same pattern works in Go, Rust, Java, Ruby, and more.
### Does streaming cost more than regular API calls?
No. Streaming and non-streaming requests cost the same — you are charged based on the number of tokens processed, not how they are delivered.
### How do I handle errors during streaming?
Implement try/catch blocks around the stream reader, use exponential backoff for retries, and always send an error event to the client so the frontend can display an appropriate message.
### Can users stop a streaming response mid-generation?
Yes. Use an AbortController (JavaScript) or cancel the async generator (Python) to stop generation. The user sees the partial response that was already streamed.
