🔹 Server-Sent Events (SSE) Explained with Python: A Simple & Elegant Tutorial

🎯 What You’ll Learn:
- What is SSE (Server-Sent Events)?
- How to build a simple SSE server in Python using FastAPI.
- How to build an SSE client to receive real-time updates.
- How SSE compares to WebSockets & Polling.

🔹 What is SSE (Server-Sent Events)?

✅ SSE (Server-Sent Events) allows a server to push updates to a client over HTTP in real-time.
✅ Unlike WebSockets, SSE is one-way (server → client).
✅ Uses HTTP Streaming to continuously send data to the client.

🔹 Best for: Real-time notifications, live dashboards, stock prices, etc.
🔹 Not for: Full-duplex (bidirectional) communication (use WebSockets for that).

📌 Step 1: Install Dependencies

Before running the code, install the required packages: sh pip install fastapi uvicorn aiohttp

📌 Step 2: Create the SSE Server (sse_server.py)

This server streams messages to clients using SSE.

from fastapi import FastAPI
from fastapi.responses import StreamingResponse
import asyncio

app = FastAPI()

async def event_stream():
    """Sends a new event every 2 seconds."""
    counter = 0
    while True:
        counter += 1
        yield f"data: Message {counter}\n\n"  # SSE format
        await asyncio.sleep(2)

@app.get("/sse")
async def sse_endpoint():
    """SSE endpoint that streams real-time events."""
    return StreamingResponse(event_stream(), media_type="text/event-stream")

if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8000)

🔹 How It Works:
- event_stream(): Generates messages every 2 seconds.
- StreamingResponse(): Streams the data to clients over HTTP.
- Sends messages in SSE format (data: message \n\n).

📌 Step 3: Create the SSE Client (sse_client.py)

This client listens for messages from the server.

import aiohttp
import asyncio

async def sse_client():
    url = "http://localhost:8000/sse"
    async with aiohttp.ClientSession() as session:
        async with session.get(url) as response:
            async for line in response.content:
                if line:
                    print(f"📩 Received from Server: {line.decode().strip()}")

asyncio.run(sse_client())

🔹 How It Works:
- aiohttp.ClientSession(): Creates an HTTP session.
- session.get(url): Opens an SSE connection.
- async for line in response.content: Listens for new messages.

📌 Step 4: Running the SSE Demo

✅ Start the Server

Run the SSE server in one terminal: sh python sse_server.py You should see: INFO: Uvicorn running on http://0.0.0.0:8000 (Press CTRL+C to quit)

✅ Start the Client

Run the client in another terminal: sh python sse_client.py You will receive: 📩 Received from Server: data: Message 1 📩 Received from Server: data: Message 2 📩 Received from Server: data: Message 3 ...

🎉 Now you have a working SSE example!

🔹 Step 5: Testing with curl

Instead of a Python client, you can use curl to listen for SSE events: sh curl -N -H "Accept: text/event-stream" http://localhost:8000/sse You’ll see: data: Message 1 data: Message 2 data: Message 3 ...

🔹 SSE vs WebSockets vs Polling

Feature SSE (Server-Sent Events) WebSockets Polling
Connection One-way (Server → Client) Two-way (Full Duplex) Repeated HTTP Requests
Use Case Live updates, notifications Chat, gaming, real-time apps Basic updates
Efficiency ✅ Efficient ✅ Efficient ❌ High Latency & Load
Browser Support ✅ Native Support ❌ Requires JS API ✅ Simple but slow

🚀 Summary

🔹 SSE is perfect for real-time server updates where the client only needs to receive data.
🔹 WebSockets are better for bidirectional communication (e.g., chat apps).
🔹 Polling is inefficient and should be avoided when real-time data is needed.

🚀 Now you fully understand Server-Sent Events! 🎯🔥
Let me know if you need more modifications or explanations! 😊🚀

TUTORIAL -2

🚀 Full SSE Demo with Docker & Streamlit Frontend

This version includes:
✅ SSE Server (FastAPI)
✅ SSE Client (Python aiohttp)
✅ Frontend (Streamlit) to display real-time events
✅ Docker Compose to containerize everything

📌 Step 1: Create the SSE Server (sse_server.py)

This FastAPI server streams messages to connected clients.

from fastapi import FastAPI
from fastapi.responses import StreamingResponse
import asyncio

app = FastAPI()

async def event_stream():
    """Sends a new event every 2 seconds."""
    counter = 0
    while True:
        counter += 1
        yield f"data: Message {counter}\n\n"  # SSE format
        await asyncio.sleep(2)

@app.get("/sse")
async def sse_endpoint():
    """SSE endpoint that streams real-time events."""
    return StreamingResponse(event_stream(), media_type="text/event-stream")

if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8000)

📌 Step 2: Create the SSE Client (sse_client.py)

This listens to the SSE server and prints received messages.

import aiohttp
import asyncio

async def sse_client():
    url = "http://sse-server:8000/sse"  # Docker container hostname
    async with aiohttp.ClientSession() as session:
        async with session.get(url) as response:
            async for line in response.content:
                if line:
                    print(f"📩 Received from Server: {line.decode().strip()}")

asyncio.run(sse_client())

📌 Step 3: Create the Streamlit Frontend (frontend.py)

This shows real-time SSE events in a web UI.

import streamlit as st
import requests
import time

st.title("📡 Real-Time SSE Events")

st.markdown("### Incoming Messages:")

# Container for messages
messages = st.empty()

def get_sse_messages():
    url = "http://sse-server:8000/sse"
    with requests.get(url, stream=True) as response:
        for line in response.iter_lines():
            if line:
                yield line.decode()

# Display messages in real-time
for msg in get_sse_messages():
    messages.text(msg)
    time.sleep(0.5)  # Smooth UI update

📌 Step 4: Containerize with Docker

✅ Create a Dockerfile for the SSE Server

# FastAPI SSE Server
FROM python:3.11-slim

WORKDIR /app
COPY sse_server.py /app/
RUN pip install fastapi uvicorn

CMD ["python", "sse_server.py"]

✅ Create a Dockerfile for the Streamlit Frontend

# Streamlit Frontend
FROM python:3.11-slim

WORKDIR /app
COPY frontend.py /app/
RUN pip install streamlit requests

CMD ["streamlit", "run", "frontend.py", "--server.port=8501", "--server.address=0.0.0.0"]

📌 Step 5: Create docker-compose.yml

This defines three services:
1️⃣ sse-server (FastAPI SSE server)
2️⃣ sse-client (Python SSE listener)
3️⃣ frontend (Streamlit UI to display messages) 

version: '3.8'

services:
  sse-server:
    build: .
    container_name: sse-server
    ports:
      - "8000:8000"

  sse-client:
    build: .
    container_name: sse-client
    depends_on:
      - sse-server
    entrypoint: ["python"]
    command: ["sse_client.py"]

  frontend:
    build: .
    container_name: sse-frontend
    depends_on:
      - sse-server
    ports:
      - "8501:8501"
    entrypoint: ["python"]
    command: ["frontend.py"]

📌 Step 6: Run the Full Setup

✅ Build and Start the Containers

docker-compose up --build

✅ This will: - Start the SSE Server on http://localhost:8000/sse - Run the SSE Client to listen for messages - Launch the Streamlit UI at http://localhost:8501

📌 Step 7: View the Live Events

1️⃣ Check the logs for the SSE Client
sh docker logs -f sse-client 🟢 Output: 📩 Received from Server: data: Message 1 📩 Received from Server: data: Message 2 ...

2️⃣ Visit the Frontend:
📌 Go to http://localhost:8501
🔹 You will see messages appearing in real-time!

🚀 Final Takeaway

✅ SSE Server: Pushes real-time messages.
✅ SSE Client: Listens & prints updates.
✅ Frontend UI: Displays messages live using Streamlit.
✅ Dockerized & scalable!

🚀 Now you have a fully working, containerized SSE example! 🎯🔥
Let me know if you need refinements! 😊🚀

TUTORIAL 3

🚀 Fully Interactive SSE System with Input & Acknowledgment

This upgrade includes:
✅ SSE Server (FastAPI) with JSON Input Handling
✅ Streamlit Frontend with a Button & Text Input for Sending Messages
✅ Dockerized Setup for Easy Deployment

📌 Step 1: Update the SSE Server (sse_server.py)

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

app = FastAPI()

message_queue = asyncio.Queue()  # Stores messages from clients

async def event_stream():
    """Sends events, including user acknowledgments."""
    while True:
        message = await message_queue.get()  # Wait for a new message
        yield f"data: Acknowledgment - Received: {message}\n\n"  # SSE format

@app.get("/sse")
async def sse_endpoint():
    """SSE endpoint for real-time streaming."""
    return StreamingResponse(event_stream(), media_type="text/event-stream")

@app.post("/messages")
async def post_message(request: Request):
    """Receives JSON messages and adds them to the queue."""
    data = await request.json()
    message = data.get("message", "No message provided")
    await message_queue.put(json.dumps({"status": "Received", "message": message}))
    return {"status": "Message received", "message": message}

if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8000)

✅ New Feature: Messages sent via POST /messages are acknowledged in the SSE stream.

📌 Step 2: Update the Streamlit Frontend (frontend.py)

import streamlit as st
import requests
import threading

st.title("📡 Real-Time SSE with Acknowledgments")

st.markdown("### Incoming Messages:")
messages_container = st.empty()

# Function to listen for SSE events
def get_sse_messages():
    url = "http://sse-server:8000/sse"
    with requests.get(url, stream=True) as response:
        for line in response.iter_lines():
            if line:
                messages_container.text(line.decode())

# Start SSE listener in a separate thread (to keep UI responsive)
thread = threading.Thread(target=get_sse_messages, daemon=True)
thread.start()

st.markdown("### Send a Message to the Server:")
user_input = st.text_input("Enter your message:")
if st.button("Send"):
    response = requests.post("http://sse-server:8000/messages", json={"message": user_input})
    st.success(f"Server Response: {response.json()}")

✅ New Feature: Sends messages via POST, and SSE stream confirms receipt.

📌 Step 3: Update docker-compose.yml

Adds three services: 1️⃣ sse-server → FastAPI SSE backend
2️⃣ sse-client → Background Python SSE listener
3️⃣ frontend → Streamlit UI for sending messages & receiving responses 

version: '3.8'

services:
  sse-server:
    build: .
    container_name: sse-server
    ports:
      - "8000:8000"

  sse-client:
    build: .
    container_name: sse-client
    depends_on:
      - sse-server
    entrypoint: ["python"]
    command: ["sse_client.py"]

  frontend:
    build: .
    container_name: sse-frontend
    depends_on:
      - sse-server
    ports:
      - "8501:8501"
    entrypoint: ["python"]
    command: ["frontend.py"]

📌 Step 4: Running the Full System

✅ Build & Start the Containers

docker-compose up --build

✅ This starts: - SSE Server (http://localhost:8000/sse) - Frontend UI (http://localhost:8501)

📌 Step 5: Testing

1️⃣ Go to http://localhost:8501
2️⃣ Enter a message & click Send
3️⃣ Check the "Incoming Messages" section → You’ll see:
data: Acknowledgment - Received: {"status": "Received", "message": "Hello Server"}

🚀 Final Takeaway

✅ Full interactive SSE system
✅ Messages sent via Streamlit UI get acknowledged live
✅ Dockerized for easy deployment

🚀 Now you have a fully working two-way SSE system! 🎯🔥
Let me know if you need further modifications! 😊🚀

REST vs SSE

🔹 REST vs SSE (Server-Sent Events) – Key Differences

✅ REST (Representational State Transfer) and SSE (Server-Sent Events) are both HTTP-based communication methods, but they serve different purposes.

🔹 1. Overview

Feature REST (HTTP APIs) SSE (Server-Sent Events)
Type of Communication Request-Response One-Way Streaming (Server → Client)
Initiated By Client Server
Use Case CRUD operations (Create, Read, Update, Delete) Real-time notifications, live updates
Connection Type Short-lived Persistent (long-lived)
Efficiency Higher latency due to repeated requests Low latency with efficient updates
State Management Stateless Stateful connection
Example Fetch user details, submit forms Live stock updates, real-time chat notifications

🔹 2. How They Work

✅ REST (Traditional APIs)

1️⃣ Client sends a request (e.g., GET /users/123)
2️⃣ Server processes the request
3️⃣ Server sends a response (e.g., user details in JSON format)
4️⃣ Connection closes after response

🔹 REST Example: http GET /users/123 HTTP/1.1 Host: api.example.com 🔹 Response: json { "id": 123, "name": "John Doe", "email": "john@example.com" }

✅ Best for: Fetching and modifying data on demand.
❌ Not good for: Real-time updates, as the client must continuously poll the server.

✅ SSE (Real-Time Data Streaming)

1️⃣ Client makes an initial connection (GET /sse).
2️⃣ Server keeps the connection open and pushes updates as they happen.
3️⃣ Client passively listens and receives messages without making new requests.

🔹 SSE Example: http GET /sse HTTP/1.1 Host: api.example.com Accept: text/event-stream 🔹 Server Response (Continuous Messages): ``` data: {"message": "Stock price updated: $120"}

data: {"message": "Breaking news: New product launched!"} ```

✅ Best for: Live feeds, notifications, financial updates.
❌ Not good for: Two-way communication (use WebSockets instead).

🔹 3. When to Use REST vs SSE?

Scenario Use REST? Use SSE?
Fetching data once ✅ Yes ❌ No
Submitting a form ✅ Yes ❌ No
Live notifications ❌ No ✅ Yes
Real-time stock prices ❌ No ✅ Yes
IoT sensor updates ❌ No ✅ Yes
Chat application ❌ No ❌ No (Use WebSockets)

🚀 Final Takeaway

Feature REST SSE
Best for Standard API interactions Real-time updates
Efficiency Less efficient for real-time Optimized for live data
Scalability Easy to scale Can require extra load balancing
Alternative WebSockets for bidirectional communication WebSockets for full-duplex communication

🚀 Now you fully understand when to use REST and when to use SSE! 🎯🔥
Let me know if you need more details! 😊🚀

🚀 Adding REST Endpoint & "Send as REST" Button in Frontend

This update includes:
✅ A RESTful API endpoint (/send_rest) to send messages
✅ A "Send as REST" button in the frontend
✅ Backend acknowledges REST messages and pushes them to SSE clients

📌 Step 1: Update the SSE Server (sse_server.py)

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

app = FastAPI()

message_queue = asyncio.Queue()  # Stores messages from clients

async def event_stream():
    """Streams messages, including REST and UI acknowledgments."""
    while True:
        message = await message_queue.get()  # Wait for a new message
        yield f"data: {message}\n\n"  # SSE format

@app.get("/sse")
async def sse_endpoint():
    """SSE endpoint for real-time streaming."""
    return StreamingResponse(event_stream(), media_type="text/event-stream")

@app.post("/messages")
async def post_message(request: Request):
    """Receives messages from the UI and adds them to the SSE queue."""
    data = await request.json()
    message = data.get("message", "No message provided")
    response = {"status": "Received", "message": message}
    await message_queue.put(json.dumps(response))
    return response

@app.post("/send_rest")
async def send_rest_message(request: Request):
    """Receives messages from REST API and adds them to SSE queue."""
    data = await request.json()
    message = data.get("message", "No REST message provided")
    response = {"status": "Received via REST", "message": message}
    await message_queue.put(json.dumps(response))  # Send to SSE
    return response

if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8000)

✅ New Feature: Messages sent via POST /send_rest are acknowledged in the SSE stream.

📌 Step 2: Update the Streamlit Frontend (frontend.py)

import streamlit as st
import requests
import threading

st.title("📡 Real-Time SSE with REST Integration")

st.markdown("### Incoming Messages:")
messages_container = st.empty()

# Function to listen for SSE events
def get_sse_messages():
    url = "http://sse-server:8000/sse"
    with requests.get(url, stream=True) as response:
        for line in response.iter_lines():
            if line:
                messages_container.text(line.decode())

# Start SSE listener in a separate thread
thread = threading.Thread(target=get_sse_messages, daemon=True)
thread.start()

st.markdown("### Send a Message to the Server:")
user_input = st.text_input("Enter your message:")

col1, col2 = st.columns(2)

# Button to send message via SSE
if col1.button("Send as SSE"):
    response = requests.post("http://sse-server:8000/messages", json={"message": user_input})
    st.success(f"SSE Response: {response.json()}")

# Button to send message via REST API
if col2.button("Send as REST"):
    response = requests.post("http://sse-server:8000/send_rest", json={"message": user_input})
    st.success(f"REST Response: {response.json()}")

✅ New Feature: "Send as REST" button sends a message via REST API (POST /send_rest) and receives SSE acknowledgment.

📌 Step 3: Update docker-compose.yml

version: '3.8'

services:
  sse-server:
    build: .
    container_name: sse-server
    ports:
      - "8000:8000"

  sse-client:
    build: .
    container_name: sse-client
    depends_on:
      - sse-server
    entrypoint: ["python"]
    command: ["sse_client.py"]

  frontend:
    build: .
    container_name: sse-frontend
    depends_on:
      - sse-server
    ports:
      - "8501:8501"
    entrypoint: ["python"]
    command: ["frontend.py"]

📌 Step 4: Running the System

✅ Build & Start the Containers

docker-compose up --build

✅ This starts: - SSE Server (http://localhost:8000/sse) - Streamlit Frontend (http://localhost:8501)

📌 Step 5: Testing

1️⃣ Go to http://localhost:8501
2️⃣ Enter a message & click "Send as SSE"
3️⃣ Enter another message & click "Send as REST"
4️⃣ Check the Incoming Messages section → You’ll see: data: {"status": "Received", "message": "Hello from SSE"} data: {"status": "Received via REST", "message": "Hello from REST"}

🚀 Final Takeaway

✅ SSE Server now supports both SSE & REST API!
✅ Streamlit UI allows sending messages via both methods!
✅ REST API messages are acknowledged via SSE stream!
✅ Fully containerized & scalable!

🚀 Now you have a complete real-time SSE system with REST support! 🎯🔥
Let me know if you need any refinements! 😊🚀

REST AMENDED

🔹 Should REST Respond Directly Instead of Using SSE Queue?

✅ Yes, a standard REST API should respond immediately to the request, instead of relying on the SSE queue.
✅ However, if we want REST messages to also notify SSE clients, we need to both return a direct response and push the message into the SSE queue.

📌 Correcting the REST Endpoint (sse_server.py)

✅ REST /send_rest now returns an immediate response.
✅ It still pushes the message into the SSE queue for real-time updates.

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

app = FastAPI()

message_queue = asyncio.Queue()  # Stores messages for SSE clients

async def event_stream():
    """Streams messages in real-time to SSE clients."""
    while True:
        message = await message_queue.get()
        yield f"data: {message}\n\n"  # SSE format

@app.get("/sse")
async def sse_endpoint():
    """SSE endpoint for real-time streaming."""
    return StreamingResponse(event_stream(), media_type="text/event-stream")

@app.post("/messages")
async def post_message(request: Request):
    """Receives messages from UI and adds them to the SSE queue."""
    data = await request.json()
    message = data.get("message", "No message provided")
    response = {"status": "Received", "message": message}
    
    # ✅ SSE clients get notified
    await message_queue.put(json.dumps(response))  
    
    return response  # ✅ REST immediately returns a response

@app.post("/send_rest")
async def send_rest_message(request: Request):
    """Receives REST messages, responds immediately, and notifies SSE clients."""
    data = await request.json()
    message = data.get("message", "No REST message provided")
    
    response = {"status": "Received via REST", "message": message}
    
    # ✅ Immediate REST response
    immediate_response = response.copy()
    
    # ✅ Push to SSE queue so real-time clients get notified
    await message_queue.put(json.dumps(response))  

    return immediate_response  # ✅ REST returns response instantly

📌 What’s Fixed?

Feature Before (❌ Bad) Now (✅ Fixed)
REST /send_rest response Delayed until SSE processed Returns immediately
SSE Notifications for REST messages ❌ Not always sent ✅ Now also sent to SSE clients

✅ Now, REST clients get an instant response, while SSE clients still receive real-time updates!

📌 Corrected Streamlit Frontend (frontend.py)

import streamlit as st
import requests
import threading

st.title("📡 Real-Time SSE with REST Support")

st.markdown("### Incoming Messages:")
messages_container = st.empty()

# Function to listen for SSE events
def get_sse_messages():
    url = "http://sse-server:8000/sse"
    with requests.get(url, stream=True) as response:
        for line in response.iter_lines():
            if line:
                messages_container.text(line.decode())

# Start SSE listener in a separate thread
thread = threading.Thread(target=get_sse_messages, daemon=True)
thread.start()

st.markdown("### Send a Message to the Server:")
user_input = st.text_input("Enter your message:")

col1, col2 = st.columns(2)

if col1.button("Send as SSE"):
    response = requests.post("http://sse-server:8000/messages", json={"message": user_input})
    st.success(f"SSE Response: {response.json()}")

if col2.button("Send as REST"):
    response = requests.post("http://sse-server:8000/send_rest", json={"message": user_input})
    st.success(f"REST Response: {response.json()}")  # ✅ Shows immediate REST response

📌 Final Testing

✅ Test REST API (Direct Response)

curl -X POST -H "Content-Type: application/json" -d '{"message": "REST says hello"}' http://localhost:8000/send_rest

Expected Immediate Response: json { "status": "Received via REST", "message": "REST says hello" }

✅ Check Live SSE Stream

curl -N -H "Accept: text/event-stream" http://localhost:8000/sse

Expected SSE Event: data: {"status": "Received via REST", "message": "REST says hello"}

🚀 Final Takeaway

✅ REST API now returns an instant response.
✅ SSE clients still receive live updates from REST messages.
✅ Perfect balance between synchronous & real-time updates!

🚀 Now you have a fully working system with instant REST responses + SSE live updates! 🎯🔥
Let me know if you need any refinements! 😊🚀

🚀 Transform your business with cutting-edge AI chatbot solutions and expert consultancy! 🌟 Contact us today at https://www.schogini.com and take the first step toward smarter automation and unparalleled growth!