chriscarrollsmith/sse-demo.py

## sse-demo.py
from fastapi import FastAPI, Request
from fastapi.responses import HTMLResponse, StreamingResponse
import time

app = FastAPI()

index_html = """
<html>
<head>
    <!-- Include htmx -->
    <script src="https://unpkg.com/htmx.org@2.0.4"></script>
    <!-- Include the SSE extension for htmx -->
    <script src="https://unpkg.com/htmx-ext-sse@2.2.2/sse.js"></script>
</head>
<body>
    <h1>HTMX + SSE Minimal Demo</h1>

    <!-- We install the SSE extension on this parent <div>
         and point sse-connect to our /sse endpoint -->
    <div id="chatContainer" hx-ext="sse" hx-swap="beforeend" sse-connect="/sse" sse-swap="newMessage" sse-close="endStream">

        <!--
            Each child <div> can listen to a different event type using sse-swap="<eventName>".
         -->

    </div>

</body>
</html>
"""

message_html = (
    "<div id=\"messageContainer\" hx-swap=\"beforeend\" "
    "sse-swap={token_event_name} style=\"border: 1px solid #ccc; padding: "
    "10px; margin: 10px 0; border-radius: 5px; background-color: #f9f9f9; "
    "display: block; width: fit-content;\"></div>"
)

@app.get("/", response_class=HTMLResponse)
def index():
    """Serve the minimal demo HTML page."""
    return index_html

@app.get("/sse")
def sse_endpoint(request: Request):
    """
    Return an SSE stream that alternates between
    - creating a new 'assistantMessage' block, and
    - sending token updates for that block.
    """
    def event_generator(request: Request):
        # Create first message
        message = message_html.format(token_event_name="newToken1")
        yield "event: newMessage\ndata:" + message + "\n\n"
        time.sleep(1)

        # Send a few tokens for first message
        for i in range(5):
            yield f"event: newToken1\ndata: Token{i} \n\n"
            time.sleep(0.5)

        # Create second message
        message = message_html.format(token_event_name="newToken2")
        yield "event: newMessage\ndata:" + message + "\n\n"
        time.sleep(1)

        # Send a few tokens for second message
        for i in range(5):
            yield f"event: newToken2\ndata: Token{i} \n\n"
            time.sleep(0.5)

        yield "event: endStream\ndata: \n\n"

    # Wrap our generator in a StreamingResponse with SSE media type
    return StreamingResponse(
        event_generator(request),
        media_type="text/event-stream"
    )
	from fastapi import FastAPI, Request
	from fastapi.responses import HTMLResponse, StreamingResponse
	import time

	app = FastAPI()

	index_html = """
	<html>
	<head>
	<!-- Include htmx -->
	<script src="https://unpkg.com/htmx.org@2.0.4"></script>
	<!-- Include the SSE extension for htmx -->
	<script src="https://unpkg.com/htmx-ext-sse@2.2.2/sse.js"></script>
	</head>
	<body>
	<h1>HTMX + SSE Minimal Demo</h1>

	<!-- We install the SSE extension on this parent <div>
	and point sse-connect to our /sse endpoint -->
	<div id="chatContainer" hx-ext="sse" hx-swap="beforeend" sse-connect="/sse" sse-swap="newMessage" sse-close="endStream">

	<!--
	Each child <div> can listen to a different event type using sse-swap="<eventName>".
	-->

	</div>

	</body>
	</html>
	"""

	message_html = (
	"<div id=\"messageContainer\" hx-swap=\"beforeend\" "
	"sse-swap={token_event_name} style=\"border: 1px solid #ccc; padding: "
	"10px; margin: 10px 0; border-radius: 5px; background-color: #f9f9f9; "
	"display: block; width: fit-content;\"></div>"
	)

	@app.get("/", response_class=HTMLResponse)
	def index():
	"""Serve the minimal demo HTML page."""
	return index_html

	@app.get("/sse")
	def sse_endpoint(request: Request):
	"""
	Return an SSE stream that alternates between
	- creating a new 'assistantMessage' block, and
	- sending token updates for that block.
	"""
	def event_generator(request: Request):
	# Create first message
	message = message_html.format(token_event_name="newToken1")
	yield "event: newMessage\ndata:" + message + "\n\n"
	time.sleep(1)

	# Send a few tokens for first message
	for i in range(5):
	yield f"event: newToken1\ndata: Token{i} \n\n"
	time.sleep(0.5)

	# Create second message
	message = message_html.format(token_event_name="newToken2")
	yield "event: newMessage\ndata:" + message + "\n\n"
	time.sleep(1)

	# Send a few tokens for second message
	for i in range(5):
	yield f"event: newToken2\ndata: Token{i} \n\n"
	time.sleep(0.5)

	yield "event: endStream\ndata: \n\n"

	# Wrap our generator in a StreamingResponse with SSE media type
	return StreamingResponse(
	event_generator(request),
	media_type="text/event-stream"
	)