PythonでServer-Sent Events

uv init -p 3.12.9 sse-server && cd sse-server

uv add fastapi uvicorn

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

app = FastAPI()

@app.post("/echo")
async def echo(request: Request):
    body = await request.body()
    text = body.decode("utf-8")

    async def sse_stream():
        for t in text:
          yield f"data: {t}\n\n"
          await asyncio.sleep(0.1)

    return StreamingResponse(
        sse_stream(),
        media_type="text/event-stream"
    )

uv run uvicorn main:app --reload

curl -N -X POST http://127.0.0.1:8000/echo -d "やっほー。明日の天気はどんな感じ？"

data: や

data: っ

data: ほ

data: ー

data: 。

data: 明

data: 日

data: の

data: 天

data: 気

data: は

data: ど

data: ん

data: な

data: 感

data: じ

data: ？

uv init -p 3.12.9 sse-client && cd sse-client
uv add sseclient-py requests

import requests
from sseclient import SSEClient

def get_stream(url: str, message: str):
    headers = {
        "Content-Type": "text/plain",
        "Accept": "text/event-stream",
    }

    response = requests.post(
        url,
        data=message.encode("utf-8"),
        headers=headers,
        stream=True
    )

    client = SSEClient(response)

    for event in client.events():
        print("received: ", event.data)

if __name__ == "__main__":
    get_stream(
        "http://127.0.0.1:8000/echo",
        "やっほー。明日の天気はどんな感じ？"
    )

uv run client.py

received:  や
received:  っ
received:  ほ
received:  ー
received:  。
received:  明
received:  日
received:  の
received:  天
received:  気
received:  は
received:  ど
received:  ん
received:  な
received:  感
received:  じ
received:  ？