Building a Chat Widget from Scratch: Frontend to Backend Complete Tutorial

Why Build Your Own Chat Widget

Third-party chat widgets give you a quick start, but they lock you into someone else's data model, rate limits, and pricing tiers. Building your own gives you full control over the user experience, data pipeline, and agent behavior. More importantly, when your chat agent needs to call internal APIs, query proprietary databases, or enforce custom business rules, an owned widget is the only architecture that scales.

This tutorial walks through building a chat widget with a React frontend and a FastAPI WebSocket backend. By the end, you will have a working system where users type messages, the backend processes them through an AI agent, and responses stream back in real time.

The Backend: FastAPI WebSocket Server

Start with the WebSocket server. FastAPI makes WebSocket handling straightforward with its native support:

flowchart LR
    CORPUS[("Pre-training corpus<br/>trillions of tokens")]
    FILTER["Quality filter and<br/>dedupe"]
    TOK["BPE tokenizer"]
    SHARD["Shard plus<br/>data parallel"]
    GPU{"GPU cluster<br/>FSDP or DeepSpeed"}
    CKPT[("Checkpoints<br/>every N steps")]
    LOSS["Loss curve plus<br/>eval gates"]
    SFT["SFT phase"]
    DPO["DPO or RLHF"]
    BASE([Base model])
    INSTR([Instruct model])
    CORPUS --> FILTER --> TOK --> SHARD --> GPU
    GPU --> CKPT --> LOSS
    LOSS --> BASE --> SFT --> DPO --> INSTR
    style GPU fill:#4f46e5,stroke:#4338ca,color:#fff
    style LOSS fill:#f59e0b,stroke:#d97706,color:#1f2937
    style INSTR fill:#059669,stroke:#047857,color:#fff

from fastapi import FastAPI, WebSocket, WebSocketDisconnect
from datetime import datetime
import json
import uuid

app = FastAPI()

class ConnectionManager:
    def __init__(self):
        self.active_connections: dict[str, WebSocket] = {}

    async def connect(self, session_id: str, websocket: WebSocket):
        await websocket.accept()
        self.active_connections[session_id] = websocket

    def disconnect(self, session_id: str):
        self.active_connections.pop(session_id, None)

    async def send_message(self, session_id: str, message: dict):
        ws = self.active_connections.get(session_id)
        if ws:
            await ws.send_json(message)

manager = ConnectionManager()

@app.websocket("/ws/chat/{session_id}")
async def chat_endpoint(websocket: WebSocket, session_id: str):
    await manager.connect(session_id, websocket)
    try:
        while True:
            data = await websocket.receive_json()
            user_message = data.get("content", "")

            # Acknowledge receipt
            await manager.send_message(session_id, {
                "type": "typing",
                "status": True,
            })

            # Process through AI agent
            response = await process_with_agent(user_message, session_id)

            await manager.send_message(session_id, {
                "type": "message",
                "id": str(uuid.uuid4()),
                "role": "assistant",
                "content": response,
                "timestamp": datetime.utcnow().isoformat(),
            })

            await manager.send_message(session_id, {
                "type": "typing",
                "status": False,
            })
    except WebSocketDisconnect:
        manager.disconnect(session_id)

The ConnectionManager tracks active WebSocket connections by session ID, allowing you to route messages to the correct client. Each incoming message triggers a typing indicator, processes through your agent, and sends the response back.

The Frontend: React Chat Component

The React component manages the WebSocket lifecycle, renders messages, and handles user input:

import { useState, useEffect, useRef, useCallback } from "react";

interface Message {
  id: string;
  role: "user" | "assistant";
  content: string;
  timestamp: string;
}

export function ChatWidget({ sessionId }: { sessionId: string }) {
  const [messages, setMessages] = useState<Message[]>([]);
  const [input, setInput] = useState("");
  const [isTyping, setIsTyping] = useState(false);
  const wsRef = useRef<WebSocket | null>(null);
  const messagesEndRef = useRef<HTMLDivElement>(null);

  useEffect(() => {
    const ws = new WebSocket(
      `wss://api.example.com/ws/chat/${sessionId}`
    );

    ws.onmessage = (event) => {
      const data = JSON.parse(event.data);
      if (data.type === "typing") {
        setIsTyping(data.status);
      } else if (data.type === "message") {
        setMessages((prev) => [...prev, data]);
      }
    };

    ws.onclose = () => {
      setTimeout(() => ws.close(), 3000); // Reconnect logic
    };

    wsRef.current = ws;
    return () => ws.close();
  }, [sessionId]);

  const sendMessage = useCallback(() => {
    if (!input.trim() || !wsRef.current) return;

    const userMsg: Message = {
      id: crypto.randomUUID(),
      role: "user",
      content: input.trim(),
      timestamp: new Date().toISOString(),
    };

    setMessages((prev) => [...prev, userMsg]);
    wsRef.current.send(JSON.stringify({ content: input.trim() }));
    setInput("");
  }, [input]);

  useEffect(() => {
    messagesEndRef.current?.scrollIntoView({ behavior: "smooth" });
  }, [messages, isTyping]);

  return (
    <div className="chat-widget">
      <div className="messages">
        {messages.map((msg) => (
          <div key={msg.id} className={`message ${msg.role}`}>
            {msg.content}
          </div>
        ))}
        {isTyping && <div className="typing-indicator">Agent is typing...</div>}
        <div ref={messagesEndRef} />
      </div>
      <div className="input-area">
        <input
          value={input}
          onChange={(e) => setInput(e.target.value)}
          onKeyDown={(e) => e.key === "Enter" && sendMessage()}
          placeholder="Type your message..."
        />
        <button onClick={sendMessage}>Send</button>
      </div>
    </div>
  );
}

Message Persistence

Store messages in a database so conversations survive page refreshes. Add a simple persistence layer:

from sqlalchemy import Column, String, Text, DateTime
from sqlalchemy.ext.asyncio import AsyncSession

class ChatMessage(Base):
    __tablename__ = "chat_messages"

    id = Column(String(36), primary_key=True)
    session_id = Column(String(36), index=True, nullable=False)
    role = Column(String(20), nullable=False)
    content = Column(Text, nullable=False)
    created_at = Column(DateTime, default=datetime.utcnow)

async def save_message(db: AsyncSession, msg: dict):
    record = ChatMessage(**msg)
    db.add(record)
    await db.commit()

async def get_history(db: AsyncSession, session_id: str):
    result = await db.execute(
        select(ChatMessage)
        .where(ChatMessage.session_id == session_id)
        .order_by(ChatMessage.created_at)
    )
    return result.scalars().all()

Load the history when the WebSocket connects, and save each new message as it flows through. This gives users a seamless experience across sessions.

FAQ

How do I handle WebSocket reconnection gracefully?

Implement exponential backoff on the client side. Track the reconnection attempt count, multiply the delay by 2 on each failure (capping at 30 seconds), and restore the message history from the server on reconnect. Send unsent messages from a local queue after the connection is re-established.

Should I use WebSockets or Server-Sent Events for chat?

Use WebSockets when both the client and server need to send messages (bidirectional chat). Use SSE when only the server pushes data (notifications, streaming responses). For a full chat widget where users type and receive responses, WebSockets are the correct choice because they handle bidirectional communication natively.

How do I scale WebSocket connections across multiple server instances?

Use a message broker like Redis Pub/Sub. When a message arrives at one server instance, publish it to a Redis channel. All server instances subscribe to that channel and deliver messages to their locally connected clients. This decouples the connection from the processing.

---

#ChatWidget #WebSocket #React #FastAPI #RealTime #AgenticAI #LearnAI #AIEngineering