Real-time Streaming APIs: Build Chat Apps with SSE and WebSockets (2026)

August 13, 2026 · 18 min read

Users expect AI chat interfaces to feel instantaneous. Streaming responses — where tokens appear word-by-word instead of all at once — are now the standard for production chat applications. This guide covers everything you need to build real-time streaming chat with Chinese AI models through TokenEase.

Why Streaming Matters

Perceived latency drops dramatically with streaming:

• Non-streaming: 5-second wait → feels slow
• Streaming: First token in 200ms → feels instant, full response in 5s

Time-to-first-token (TTFT) is the critical metric for user experience. Chinese models via TokenEase achieve TTFT under 300ms for most queries.

Server-Sent Events (SSE) vs WebSockets

Two protocols dominate real-time AI streaming:

Implementing SSE Streaming with TokenEase

Backend (Python/FastAPI)

from fastapi import FastAPI
from fastapi.responses import StreamingResponse
import requests
import json

app = FastAPI()
TOKEN = "your_tokenease_api_key"
BASE_URL = "https://tokenease.io/v1"

@app.post("/chat/stream")
async def chat_stream(message: str, model: str = "deepseek-v4"):
    """Stream chat completions via SSE"""
    
    def generate():
        response = requests.post(
            f"{BASE_URL}/chat/completions",
            headers={"Authorization": f"Bearer {TOKEN}"},
            json={
                "model": model,
                "messages": [{"role": "user", "content": message}],
                "stream": True,  # Enable streaming
                "temperature": 0.7
            },
            stream=True  # Enable HTTP streaming
        )
        
        for line in response.iter_lines():
            if line:
                line = line.decode("utf-8")
                if line.startswith("data: "):
                    data = line[6:]  # Remove "data: " prefix
                    if data == "[DONE]":
                        break
                    try:
                        chunk = json.loads(data)
                        delta = chunk["choices"][0]["delta"]
                        if "content" in delta:
                            token = delta["content"]
                            # SSE format: data: \n\n
                            yield f"data: {json.dumps({'token': token})}\n\n"
                    except:
                        pass
    
    return StreamingResponse(
        generate(),
        media_type="text/event-stream",
        headers={
            "Cache-Control": "no-cache",
            "Connection": "keep-alive",
            "X-Accel-Buffering": "no"  # Disable nginx buffering
        }
    )

Frontend (JavaScript)

class StreamingChat {
    constructor(apiUrl) {
        this.apiUrl = apiUrl;
        this.eventSource = null;
    }
    
    async sendMessage(message, onToken, onComplete, onError) {
        // Use fetch with ReadableStream for POST-based SSE
        const response = await fetch(this.apiUrl, {
            method: "POST",
            headers: {"Content-Type": "application/json"},
            body: JSON.stringify({message, model: "deepseek-v4"})
        });
        
        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 lines = buffer.split("\n");
            buffer = lines.pop(); // Keep incomplete line in buffer
            
            for (const line of lines) {
                if (line.startsWith("data: ")) {
                    const data = line.slice(6);
                    if (data === "[DONE]") {
                        onComplete();
                        return;
                    }
                    try {
                        const parsed = JSON.parse(data);
                        if (parsed.token) {
                            onToken(parsed.token);
                        }
                    } catch (e) {
                        // Ignore parse errors
                    }
                }
            }
        }
        onComplete();
    }
}

// Usage
const chat = new StreamingChat("/chat/stream");
const outputDiv = document.getElementById("output");

chat.sendMessage(
    "Explain quantum computing in simple terms",
    (token) => { outputDiv.textContent += token; },
    () => { console.log("Stream complete"); },
    (err) => { console.error("Error:", err); }
);

WebSocket Implementation for Multi-User Chat

# FastAPI WebSocket endpoint
from fastapi import FastAPI, WebSocket
from fastapi.websockets import WebSocketDisconnect
import asyncio

app = FastAPI()

class ConnectionManager:
    def __init__(self):
        self.active_connections = []
    
    async def connect(self, websocket):
        await websocket.accept()
        self.active_connections.append(websocket)
    
    def disconnect(self, websocket):
        self.active_connections.remove(websocket)
    
    async def broadcast(self, message):
        for connection in self.active_connections:
            await connection.send_text(message)

manager = ConnectionManager()

@app.websocket("/ws/chat")
async def websocket_chat(websocket: WebSocket):
    await manager.connect(websocket)
    try:
        while True:
            data = await websocket.receive_text()
            message = json.loads(data)
            
            # Stream AI response to all connected clients
            response = requests.post(
                f"{BASE_URL}/chat/completions",
                headers={"Authorization": f"Bearer {TOKEN}"},
                json={
                    "model": "deepseek-v4",
                    "messages": [{"role": "user", "content": message["text"]}],
                    "stream": True
                },
                stream=True
            )
            
            for line in response.iter_lines():
                if line:
                    line = line.decode("utf-8")
                    if line.startswith("data: "):
                        data_str = line[6:]
                        if data_str == "[DONE]":
                            await manager.broadcast(json.dumps({"done": True}))
                            break
                        try:
                            chunk = json.loads(data_str)
                            delta = chunk["choices"][0]["delta"]
                            if "content" in delta:
                                await manager.broadcast(json.dumps({
                                    "token": delta["content"],
                                    "user": message.get("user", "AI")
                                }))
                        except:
                            pass
                            
    except WebSocketDisconnect:
        manager.disconnect(websocket)

# Frontend WebSocket client
const ws = new WebSocket("wss://yourdomain.com/ws/chat");

ws.onmessage = (event) => {
    const data = JSON.parse(event.data);
    if (data.token) {
        appendToken(data.token, data.user);
    }
    if (data.done) {
        showMessageComplete();
    }
};

function sendMessage(text) {
    ws.send(JSON.stringify({text, user: currentUser}));
}

Latency Optimization Techniques

1. Connection Pooling

import requests
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry

# Create a session with connection pooling
session = requests.Session()
adapter = HTTPAdapter(
    pool_connections=20,
    pool_maxsize=100,
    max_retries=Retry(total=3, backoff_factor=0.5)
)
session.mount("https://", adapter)

# Reuse session for all requests
response = session.post(
    f"{BASE_URL}/chat/completions",
    headers={"Authorization": f"Bearer {TOKEN}"},
    json={"model": "deepseek-v4", "messages": messages, "stream": True},
    stream=True
)

2. Model Selection by Latency Requirements

def select_model_for_latency(requirement):
    """Select optimal model based on latency requirements"""
    models = {
        "ultra_low": "doubao-pro",      # 150ms TTFT
        "low": "kimi-k2",               # 180ms TTFT
        "balanced": "deepseek-v4",      # 250ms TTFT
        "quality": "glm-4"              # 300ms TTFT
    }
    return models.get(requirement, "deepseek-v4")

# Route simple queries to faster models
if is_simple_query(user_message):
    model = "kimi-k2"  # Faster, cheaper
else:
    model = "deepseek-v4"  # Higher quality

3. Pre-warming Connections

import asyncio

async def prewarm_connections():
    """Send warm-up requests to keep connections alive"""
    while True:
        try:
            requests.post(
                f"{BASE_URL}/chat/completions",
                headers={"Authorization": f"Bearer {TOKEN}"},
                json={
                    "model": "deepseek-v4",
                    "messages": [{"role": "user", "content": "hi"}],
                    "max_tokens": 1
                },
                timeout=5
            )
        except:
            pass
        await asyncio.sleep(30)  # Every 30 seconds

# Run in background
asyncio.create_task(prewarm_connections())

Handling Stream Interruptions

class ResilientStream {
    constructor(apiUrl, maxRetries = 3) {
        this.apiUrl = apiUrl;
        this.maxRetries = maxRetries;
        this.receivedTokens = [];
    }
    
    async sendMessage(message, onToken, onComplete) {
        let retries = 0;
        
        while (retries < this.maxRetries) {
            try {
                await this._stream(message, onToken);
                onComplete();
                return;
            } catch (error) {
                retries++;
                console.warn(`Stream failed, retry ${retries}/${this.maxRetries}`);
                
                if (retries < this.maxRetries) {
                    // Resume from last received token
                    const partialResponse = this.receivedTokens.join("");
                    const continuationPrompt = `Continue from where you left off. You said: "${partialResponse.slice(-100)}"`;
                    message = continuationPrompt;
                    await this._delay(1000 * retries); // Exponential backoff
                }
            }
        }
        
        throw new Error("Max retries exceeded");
    }
    
    async _stream(message, onToken) {
        // ... streaming implementation
    }
    
    _delay(ms) {
        return new Promise(resolve => setTimeout(resolve, ms));
    }
}

Production Checklist

• Nginx buffering disabled — X-Accel-Buffering: no header
• Timeout configuration — proxy_read_timeout 300s for long generations
• Rate limiting — Prevent abuse with per-user token budgets
• Error handling — Graceful degradation when AI service is unavailable
• Monitoring — Track TTFT, tokens/sec, error rates per model

# Nginx configuration for streaming
location /chat/stream {
    proxy_pass http://backend;
    proxy_buffering off;
    proxy_cache off;
    proxy_read_timeout 300s;
    proxy_send_timeout 300s;
    proxy_connect_timeout 10s;
}

Cost Analysis

Streaming does not increase token cost — you pay for the same tokens, just delivered faster:

• 1K token response: Same cost whether streamed or not
• Overhead: ~5% more bandwidth due to SSE framing
• User satisfaction: 3× higher with streaming vs batch responses

Next Steps

Build your streaming chat app with TokenEase:

1. Get your free API key ($1 credit)
2. Start with SSE for single-user chat
3. Add WebSockets for collaborative features
4. Monitor TTFT and optimize model selection

For production patterns, see our guides on failover and load balancing and AI agent development.

Last updated: August 2026. Streaming latency varies by model load and region.