flaskでデータをチャンクで返したいときは stream_with_context(...)を返す

from flask import Flask, Response, stream_with_context
from flask_cors import CORS
import json
import time

app = Flask(__name__)
CORS(app)

@app.route('/chunked')
def chunked():
    def generate():
        for i in range(3):
            yield f"data: {json.dumps({'response': f'Chunk {i}'})}\n\n"
            time.sleep(1)
    return Response(stream_with_context(generate()), mimetype='text/event-stream')

if __name__ == '__main__':
    app.run(debug=True, port=5000) 

こうしておくと、

$ curl -i 127.0.0.1:5000/chunked
HTTP/1.1 200 OK
Server: Werkzeug/3.0.1 Python/3.11.5
Date: Thu, 19 Jun 2025 14:12:44 GMT
Content-Type: text/event-stream; charset=utf-8
Access-Control-Allow-Origin: *
Transfer-Encoding: chunked
Connection: close

data: {"response": "Chunk 0"}

data: {"response": "Chunk 1"}

data: {"response": "Chunk 2"}

と1秒ごとにデータが降ってくる。遥か昔にロングタイムポーリングが流行った時に少し使ったことがあるくらいだったけど、現代ではLLMのAPIがストリームでデータを返してくるので、使い所があって嬉しい。

それはそうと、戻りのレスポンスヘッダの Connection は Closeなんだっけ? と一瞬思ったけどこれは keepalive かけないよってやつか。クライアントが curl であるのでそれはそう。