How to Build a Streaming AI Chat With Server-Sent Events: Complete Tutorial

When you type a message in ChatGPT and watch the response appear word by word, you're experiencing streaming — the AI's output being delivered in real-time as it's generated, rather than waiting for the complete response before showing anything.

Streaming transforms user experience. Instead of staring at a loading spinner for 10-30 seconds while a model generates a long response, users see text appear immediately and continuously. It feels faster, more engaging, and more natural — like watching someone type.

In this tutorial, you'll learn how to build a streaming AI chat application using Server-Sent Events (SSE) and an OpenAI-compatible API. We'll cover both Node.js and Python implementations, production considerations, and cost optimization.

Why Server-Sent Events?

SSE is the standard protocol for streaming AI responses. Here's why it's the right choice:

One-directional: SSE is designed for server-to-client streaming, which is exactly what AI chat needs (the client sends a request, the server streams the response)
Built on HTTP: Works everywhere, no WebSocket complexity, no special infrastructure
Auto-reconnection: Browsers automatically reconnect if the connection drops
Simple to implement: Much less code than WebSockets for this use case
OpenAI-compatible: The OpenAI API (and all compatible providers) uses SSE for streaming

Architecture Overview

code

Browser (EventSource) → Your Backend → AI API (SSE stream)
     ← ← ← ← ← ← ← ← ← ← ← ← ← ← ←
     Server-Sent Events flow back

The browser opens an SSE connection to your backend
Your backend forwards the request to the AI API with stream: true
The AI API streams response chunks back to your backend
Your backend forwards each chunk to the browser via SSE
The browser appends each chunk to the chat display in real-time

Part 1: Node.js Implementation

Backend (Express + SSE)

javascript

const express = require('express');
const app = express();
app.use(express.json());

// POST /api/chat - streaming endpoint
app.post('/api/chat', async (req, res) => {
  const { messages, model = 'gpt-5.6-terra' } = req.body;

  // Set SSE headers
  res.writeHead(200, {
    'Content-Type': 'text/event-stream',
    'Cache-Control': 'no-cache',
    'Connection': 'keep-alive',
    'X-Accel-Buffering': 'no', // Disable Nginx buffering
  });

  try {
    // Call the AI API with streaming enabled
    const response = await fetch('https://api.qubax.ai/v1/chat/completions', {
      method: 'POST',
      headers: {
        'Content-Type': 'application/json',
        'Authorization': `Bearer ${process.env.QUBAX_API_KEY}`,
      },
      body: JSON.stringify({
        model: model,
        messages: messages,
        stream: true, // Enable streaming
        max_tokens: 2000,
      }),
    });

    // Read the streaming response
    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]') {
            res.write('data: [DONE]\n\n');
            res.end();
            return;
          }
          // Forward the SSE chunk to the client
          res.write(`data: ${data}\n\n`);
        }
      }
    }
  } catch (error) {
    console.error('Stream error:', error);
    res.write(`data: ${JSON.stringify({ error: error.message })}\n\n`);
    res.end();
  }
});

app.listen(3000, () => console.log('Server running on port 3000'));

Frontend (Browser EventSource)

javascript

// Note: EventSource only supports GET, so we use fetch with streaming
async function streamChat(messages) {
  const response = await fetch('/api/chat', {
    method: 'POST',
    headers: { 'Content-Type': 'application/json' },
    body: JSON.stringify({ messages }),
  });

  const reader = response.body.getReader();
  const decoder = new TextDecoder();
  let assistantMessage = '';
  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();

    for (const line of lines) {
      if (line.startsWith('data: ')) {
        const data = line.slice(6);
        if (data === '[DONE]') return assistantMessage;

        const parsed = JSON.parse(data);
        const content = parsed.choices?.[0]?.delta?.content || '';
        assistantMessage += content;

        // Update the UI in real-time
        document.getElementById('chat-output').innerText = assistantMessage;
      }
    }
  }
  return assistantMessage;
}

Part 2: Python Implementation

Backend (FastAPI + SSE)

python

from fastapi import FastAPI, Request
from fastapi.responses import StreamingResponse
import httpx
import json

app = FastAPI()

QUBAX_API_KEY = "your-api-key"
QUBAX_API_URL = "https://api.qubax.ai/v1/chat/completions"

@app.post("/api/chat")
async def chat(request: Request):
    body = await request.json()
    messages = body.get("messages", [])
    model = body.get("model", "gpt-5.6-terra")

    async def stream_generator():
        async with httpx.AsyncClient() as client:
            async with client.stream(
                "POST",
                QUBAX_API_URL,
                json={
                    "model": model,
                    "messages": messages,
                    "stream": True,
                    "max_tokens": 2000,
                },
                headers={
                    "Authorization": f"Bearer {QUBAX_API_KEY}",
                    "Content-Type": "application/json",
                },
                timeout=60.0,
            ) as response:
                async for line in response.aiter_lines():
                    if line.startswith("data: "):
                        data = line[6:]
                        if data == "[DONE]":
                            yield f"data: [DONE]\n\n"
                            return
                        yield f"data: {data}\n\n"

    return StreamingResponse(
        stream_generator(),
        media_type="text/event-stream",
        headers={
            "Cache-Control": "no-cache",
            "Connection": "keep-alive",
            "X-Accel-Buffering": "no",
        },
    )

Python Client (for testing or backend-to-backend)

python

import httpx
import asyncio

async def stream_chat(messages, model="gpt-5.6-terra"):
    async with httpx.AsyncClient() as client:
        async with client.stream(
            "POST",
            "https://api.qubax.ai/v1/chat/completions",
            json={
                "model": model,
                "messages": messages,
                "stream": True,
            },
            headers={
                "Authorization": f"Bearer {API_KEY}",
                "Content-Type": "application/json",
            },
        ) as response:
            async for line in response.aiter_lines():
                if line.startswith("data: "):
                    data = line[6:]
                    if data == "[DONE]":
                        break
                    chunk = json.loads(data)
                    content = chunk["choices"][0]["delta"].get("content", "")
                    print(content, end="", flush=True)

asyncio.run(stream_chat([
    {"role": "user", "content": "Explain quantum computing simply"}
]))

Part 3: Production Considerations

Error Handling and Reconnection

javascript

// Frontend reconnection logic
async function robustStreamChat(messages, maxRetries = 3) {
  let retries = 0;
  while (retries < maxRetries) {
    try {
      return await streamChat(messages);
    } catch (error) {
      retries++;
      if (retries >= maxRetries) throw error;
      const delay = Math.min(1000 * 2**retries, 10000); // Exponential backoff
      await new Promise(r => setTimeout(r, delay));
    }
  }
}

Nginx Configuration

If you're behind Nginx, you need to disable buffering for SSE endpoints:

nginx

location /api/chat {
    proxy_pass http://backend:3000;
    proxy_buffering off;
    proxy_cache off;
    proxy_set_header Connection '';
    proxy_http_version 1.1;
    chunked_transfer_encoding on;
}

Rate Limiting

Streaming endpoints need special rate limiting because connections are long-lived:

javascript

const activeStreams = new Map();

function rateLimit(req, res, next) {
  const userId = req.user?.id || req.ip;
  const active = activeStreams.get(userId) || 0;

  if (active >= 5) {
    return res.status(429).json({ error: 'Too many concurrent streams' });
  }

  activeStreams.set(userId, active + 1);
  res.on('close', () => {
    const current = activeStreams.get(userId) || 1;
    activeStreams.set(userId, current - 1);
  });

  next();
}

Part 4: Cost Optimization

Streaming doesn't change the total token cost, but it affects how you manage context. Here are key strategies:

Choose the Right Model

For high-volume chat applications, model choice is your biggest cost lever:

Model	Input ($/M tokens)	Output ($/M tokens)	Best For
GPT-5.6 Terra	$1.00 retail / $0.97 Qubax	$6.00 / $5.82	General chat
DeepSeek V4 Flash	$0.14 / $0.11	$0.28 / $0.23	High volume, budget
GLM 5.2	$0.76 / $0.16	$2.42 / $0.50	Multilingual
Claude Sonnet 5	$2.00 / $1.94	$10.00 / $9.70	Complex reasoning

For a chatbot handling 10,000 conversations/day with average 2K input + 500 output tokens:

GPT-5.6 Terra: ~$39/day on Qubax
DeepSeek V4 Flash: ~$2.55/day on Qubax
GLM 5.2: ~$5.75/day on Qubax

That's a 15x cost range for similar conversational capability. Choose wisely based on your quality requirements.

Implement Context Trimming

Don't send the entire conversation history every time. Trim old messages:

javascript

function trimContext(messages, maxTokens = 8000) {
  const systemPrompt = messages[0]; // Always keep system prompt
  const conversation = messages.slice(1);
  let totalTokens = estimateTokens(systemPrompt.content);

  // Keep most recent messages that fit
  const trimmed = [systemPrompt];
  for (let i = conversation.length - 1; i >= 0; i--) {
    const msgTokens = estimateTokens(conversation[i].content);
    if (totalTokens + msgTokens > maxTokens) break;
    trimmed.splice(1, 0, conversation[i]); // Insert after system prompt
    totalTokens += msgTokens;
  }

  return trimmed;
}

function estimateTokens(text) {
  return Math.ceil(text.length / 4); // Rough estimate
}

Use Model Routing

Route simple queries to cheaper models and complex ones to premium models:

javascript

function selectModel(userMessage) {
  const complex = /code|debug|analyze|compare|architecture|design|implement/i;
  if (complex.test(userMessage)) {
    return 'claude-opus-5'; // Premium for complex tasks
  }
  return 'deepseek-v4-flash'; // Budget for simple chat
}

Testing Your Implementation

Here's a simple test to verify streaming works:

bash

# Test with curl
curl -N -X POST http://localhost:3000/api/chat \
  -H "Content-Type: application/json" \
  -d '{"messages":[{"role":"user","content":"Count from 1 to 10 slowly"}]}' \
  --no-buffer

The -N and --no-buffer flags ensure curl displays streaming output immediately. You should see chunks arrive one at a time.

Common Pitfalls

1. Buffering Issues

If chunks arrive all at once instead of streaming, check:

Nginx: Set proxy_buffering off
Express: Set X-Accel-Buffering: no header
Cloudflare: Disable response buffering for the route

2. CORS Errors

SSE requires proper CORS headers:

javascript

res.writeHead(200, {
  'Access-Control-Allow-Origin': '*',
  'Access-Control-Allow-Headers': 'Content-Type',
  'Content-Type': 'text/event-stream',
  // ... other headers
});

3. Memory Leaks

Always clean up event listeners and close connections:

javascript

req.on('close', () => {
  // Clean up: cancel the upstream fetch, clear buffers
  controller.abort();
});

4. Timeouts

Long responses may exceed default timeouts. Set generous timeouts on both your server and the AI API client.

Putting It All Together

A complete production streaming chat involves:

Backend SSE endpoint that proxies to the AI API with stream: true
Frontend fetch with streaming that reads chunks and updates the UI
Error handling with retries and graceful degradation
Rate limiting that accounts for long-lived connections
Cost optimization through model selection and context management

The result is a chat experience that feels instant and responsive — the same quality users expect from modern AI interfaces.

Ready to build? Get your API key and start streaming → [qubax.ai/docs](https://qubax.ai/docs)

Compare models and pricing for your use case → [qubax.ai/models](https://qubax.ai/models)

FAQ

What is Server-Sent Events (SSE)?

SSE is a web protocol for streaming data from a server to a browser over HTTP. It's the standard method for streaming AI responses, used by the OpenAI API and all compatible providers including Qubax.

Do I need WebSockets for AI chat streaming?

No. SSE is simpler and better suited for AI chat because it's one-directional (server-to-client). WebSockets add unnecessary complexity for this use case. Use SSE unless you need bidirectional real-time communication.

How do I enable streaming in the API?

Add "stream": true to your chat completion request. The API will return Server-Sent Events instead of a single JSON response, with each chunk containing a partial response.

Does streaming cost more than non-streaming?

No. The total token cost is the same whether you stream or not. Streaming only affects how the response is delivered — in chunks rather than all at once.

Which model should I use for a streaming chat app?

For high-volume, cost-sensitive applications, DeepSeek V4 Flash ($0.11/M input on Qubax) or GLM 5.2 ($0.16/M input on Qubax) offer excellent value. For complex reasoning, use Claude Opus 5 or GPT-5.6 Sol. See qubax.ai/models for full pricing.

How do I handle disconnections during streaming?

Implement client-side retry logic with exponential backoff. Track which chunks were already received so you can resume from the correct position. Most browsers auto-reconnect SSE connections, but fetch-based streaming requires manual retry.

Can I stream responses in Python?

Yes. Use httpx with client.stream() and aiter_lines() to process SSE chunks asynchronously. The tutorial above includes a complete Python example.

How do I disable Nginx buffering for SSE?

Add proxy_buffering off; and proxy_cache off; to the Nginx location block for your SSE endpoint. Also set the X-Accel-Buffering: no response header in your backend.

How to Build a Streaming AI Chat With Server-Sent Events: Complete Tutorial

Why Server-Sent Events?

Architecture Overview

Part 1: Node.js Implementation

Backend (Express + SSE)

Frontend (Browser EventSource)

Part 2: Python Implementation

Backend (FastAPI + SSE)

Python Client (for testing or backend-to-backend)

Part 3: Production Considerations

Error Handling and Reconnection

Nginx Configuration

Rate Limiting

Part 4: Cost Optimization

Choose the Right Model

Implement Context Trimming

Use Model Routing

Testing Your Implementation

Common Pitfalls

1. Buffering Issues

2. CORS Errors

3. Memory Leaks

4. Timeouts

Putting It All Together

FAQ

What is Server-Sent Events (SSE)?

Do I need WebSockets for AI chat streaming?

How do I enable streaming in the API?

Does streaming cost more than non-streaming?

Which model should I use for a streaming chat app?

How do I handle disconnections during streaming?

Can I stream responses in Python?

How do I disable Nginx buffering for SSE?

Article tags

Related articles