This flask chatbot openai api tutorial builds a working chat backend in Flask that talks to the OpenAI completions endpoint, keeps per-session history, and streams tokens back to the client. You’ll end up with runnable code that you can hit with curl or a browser, not a toy snippet that breaks on the second message.
Prerequisites
- Python 3.11 or newer
- pip and a virtual environment tool
- An OpenAI API key (set as
OPENAI_API_KEYin your environment) - Familiarity with Flask routing and JSON requests
If you’ve never touched Flask, you can still follow along, but I assume you know what a route decorator does and how to read a 400 response.
Project setup
Create a directory and install dependencies. We pin major versions to avoid surprise breakages from floating minors.
mkdir flask-chat && cd flask-chat
python -m venv .venv
source .venv/bin/activate
pip install "flask==3.0.0" "openai==1.14.0" "python-dotenv==1.0.0"Drop your key in a .env file so it never lands in source control:
echo "OPENAI_API_KEY=sk-..." > .envMinimal chat endpoint
The first cut is a single POST endpoint that accepts a user message and returns the model’s reply. We use the official openai SDK, which is OpenAI-compatible and handles auth and base retries.
import os
from flask import Flask, request, jsonify
from openai import OpenAI
from dotenv import load_dotenv
load_dotenv()
client = OpenAI(api_key=os.environ["OPENAI_API_KEY"])
app = Flask(__name__)
@app.post("/chat")
def chat():
data = request.get_json()
user_msg = data.get("message", "")
if not user_msg:
return jsonify({"error": "missing message"}), 400
resp = client.chat.completions.create(
model="gpt-3.5-turbo",
messages=[{"role": "user", "content": user_msg}],
temperature=0.7,
)
reply = resp.choices[0].message.content
return jsonify({"reply": reply})Run it:
flask --app app run --port 5000Test with curl:
curl -s -X POST localhost:5000/chat \
-H 'Content-Type: application/json' \
-d '{"message":"What is a Flask blueprint?"}'Expected output:
{"reply":"A Flask blueprint is a way to organize a group of related routes and handlers into a reusable module..."}That works for one-off questions. It forgets everything immediately.
Conversation memory
Real chatbots keep context. We’ll store message lists in a process-local dict keyed by a session id sent by the client. This is fine for a single-worker dev server; for multi-process you’d swap in Redis (more on that later).
from collections import defaultdict
conversations = defaultdict(list)
@app.post("/chat")
def chat():
data = request.get_json()
sid = data.get("session_id", "default")
user_msg = data.get("message", "")
if not user_msg:
return jsonify({"error": "missing message"}), 400
conversations[sid].append({"role": "user", "content": user_msg})
resp = client.chat.completions.create(
model="gpt-3.5-turbo",
messages=conversations[sid],
temperature=0.7,
)
reply = resp.choices[0].message.content
conversations[sid].append({"role": "assistant", "content": reply})
return jsonify({"reply": reply, "history": conversations[sid]})Send a follow-up:
curl -s -X POST localhost:5000/chat \
-H 'Content-Type: application/json' \
-d '{"session_id":"u1","message":"Give me a one-line example."}'The model now sees the prior exchange because we forwarded the full list. Note the unbounded growth: in production you must truncate or summarize older turns to stay under the context window.
Managing context length
The gpt-3.5-turbo context window is 16k tokens. Appending every turn unbounded will eventually blow up with a context_length_exceeded error. A simple strategy is to keep only the last N turns:
MAX_TURNS = 10
def trim(hist):
return hist[-MAX_TURNS*2:] # each turn is user + assistant
# before calling the API:
conversations[sid] = trim(conversations[sid])For long documents, use embeddings or a summarization pass. That’s out of scope for this flask chatbot openai api tutorial, but you should plan for it before shipping.
Streaming tokens
Waiting for the full response feels sluggish. The OpenAI SDK supports streaming via stream=True. We’ll yield Server-Sent Events (SSE) to the browser or any HTTP client.
from flask import Response
import json
@app.post("/chat/stream")
def chat_stream():
data = request.get_json()
sid = data.get("session_id", "default")
user_msg = data.get("message", "")
conversations[sid].append({"role": "user", "content": user_msg})
def generate():
stream = client.chat.completions.create(
model="gpt-3.5-turbo",
messages=conversations[sid],
temperature=0.7,
stream=True,
)
collected = []
for chunk in stream:
delta = chunk.choices[0].delta.content or ""
if delta:
collected.append(delta)
yield f"data: {json.dumps({'token': delta})}\n\n"
full = "".join(collected)
conversations[sid].append({"role": "assistant", "content": full})
yield f"data: [DONE]\n\n"
return Response(generate(), mimetype="text/event-stream")Curl with -N to disable buffering:
curl -N -X POST localhost:5000/chat/stream \
-H 'Content-Type: application/json' \
-d '{"session_id":"u2","message":"Explain GIL in 3 sentences."}'You’ll see tokens arrive line by line:
data: {"token":"The"}
data: {"token":" Global"}
data: {"token":" Interpreter"}
data: {"token":" Lock"}
...
data: [DONE]The SSE format is just data: <json>\n\n. A browser EventSource expects GET, so for POST we use fetch with a streaming reader as shown below.
Error handling and timeouts
OpenAI calls fail: rate limits, network blips, bad keys. Set a client timeout and catch APIError. Never let an exception bubble into a 500 with a stack trace if you can return a clean 429/502.
from openai import APIError, Timeout
client = OpenAI(api_key=os.environ["OPENAI_API_KEY"], timeout=10.0)
@app.post("/chat")
def chat():
# ... same as before ...
try:
resp = client.chat.completions.create(...)
except Timeout:
return jsonify({"error": "upstream timeout"}), 504
except APIError as e:
code = e.status_code or 500
return jsonify({"error": "openai error", "detail": str(e)}), codeIf you want resilience beyond a single provider, an OpenAI-compatible gateway such as n4n.ai will automatically fall back when a provider is rate-limited or degraded, while keeping the same request shape. That’s useful when you don’t want to write your own retry matrix.
Serving a simple front end
A chatbot needs a box to type in. Here’s a minimal HTML page that posts to the stream endpoint and prints tokens. If you serve this from a different origin, add flask-cors; we assume same-origin for brevity.
<!doctype html>
<form id="f">
<input name="msg" placeholder="say something" />
<button>Send</button>
</form>
<pre id="out"></pre>
<script>
const sid = Math.random().toString(36).slice(2);
f.onsubmit = async (e) => {
e.preventDefault();
const msg = f.msg.value;
const res = await fetch("/chat/stream", {
method: "POST",
headers: {"Content-Type": "application/json"},
body: JSON.stringify({session_id: sid, message: msg}),
});
const reader = res.body.getReader();
const dec = new TextDecoder();
while (true) {
const {value, done} = await reader.read();
if (done) break;
const lines = dec.decode(value).split("\n\n");
for (const l of lines) {
if (l.startsWith("data: ")) {
const payload = l.slice(6);
if (payload === "[DONE]") return;
out.textContent += JSON.parse(payload).token;
}
}
}
};
</script>Serve it from a route:
@app.get("/")
def index():
return app.send_static_file("index.html")Put the HTML in static/index.html. Hit / and you have a working chat UI.
Production considerations
The in-memory conversations dict dies on restart and isn’t shared across Gunicorn workers. Use Redis or a database with a TTL. Example:
import redis, json
r = redis.Redis()
def get_hist(sid):
raw = r.get(f"chat:{sid}")
return json.loads(raw) if raw else []
def save_hist(sid, hist):
r.setex(f"chat:{sid}", 3600, json.dumps(hist))Run behind Gunicorn with multiple workers:
gunicorn -w 4 -b 0.0.0.0:5000 app:appSet OPENAI_API_KEY in the environment, not in code. For per-token cost tracking, the OpenAI response includes usage fields; log them. If you route through a gateway, per-token usage metering and provider cache-control hints are forwarded automatically, which simplifies billing code.
Finally, don’t expose /chat/stream without auth. Add a middleware that checks a bearer token or session cookie. Rate-limit per IP to avoid surprise bills.
This flask chatbot openai api tutorial gave you a stateful, streaming backend and a front end. Swap the model name for gpt-4o or any fine-tune, and you’re ready to integrate into a larger app.