--- title: "Building a Chat Widget from Scratch: Frontend to Backend Complete Tutorial" description: "Learn how to build a production-quality chat widget with a React frontend component, WebSocket backend in Python, message formatting, typing indicators, and persistent message history." canonical: https://callsphere.ai/blog/building-chat-widget-from-scratch-frontend-backend-tutorial category: "Learn Agentic AI" tags: ["Chat Widget", "WebSocket", "React", "FastAPI", "Real-Time"] author: "CallSphere Team" published: 2026-03-17T00:00:00.000Z updated: 2026-05-06T01:02:45.725Z --- # Building a Chat Widget from Scratch: Frontend to Backend Complete Tutorial > Learn how to build a production-quality chat widget with a React frontend component, WebSocket backend in Python, message formatting, typing indicators, and persistent message history. ## 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: ```mermaid flowchart LR CORPUS[("Pre-training corpus trillions of tokens")] FILTER["Quality filter and dedupe"] TOK["BPE tokenizer"] SHARD["Shard plus data parallel"] GPU{"GPU cluster FSDP or DeepSpeed"} CKPT[("Checkpoints every N steps")] LOSS["Loss curve plus 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 ``` ```python 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: ```typescript 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([]); const [input, setInput] = useState(""); const [isTyping, setIsTyping] = useState(false); const wsRef = useRef(null); const messagesEndRef = useRef(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 ( {messages.map((msg) => ( {msg.content} ))} {isTyping && Agent is typing...} setInput(e.target.value)} onKeyDown={(e) => e.key === "Enter" && sendMessage()} placeholder="Type your message..." /> Send ); } ``` ## Message Persistence Store messages in a database so conversations survive page refreshes. Add a simple persistence layer: ```python 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 --- Source: https://callsphere.ai/blog/building-chat-widget-from-scratch-frontend-backend-tutorial