Webアプリケーションにおいて、サーバーからクライアントへリアルタイムにデータをプッシュする必要がある場面は多々存在します。チャットアプリケーション、ライブスポーツのスコア更新、株価ティッカー、通知システムなどがその典型例です。Server-Sent Events (SSE) は、このような単方向のサーバーからクライアントへのデータプッシュを実現するための効率的な技術です。
Server-Sent Events (SSE) の概要
SSEはHTML5の一部として標準化された仕様で、クライアント（通常はブラウザ）が一度接続を確立すると、サーバーはその接続を介して継続的にデータを送信できます。これは従来のHTTPの「リクエスト-レスポンス」モデルとは異なり、サーバーが能動的に情報を送り続けることができる点が特徴です。
厳密には、HTTPプロトコル自体がサーバーからの「プッシュ」を直接サポートしているわけではありません。しかし、SSEはこの制約を「ストリーミング」という形で回避します。クライアントがサーバーにHTTPリクエストを送信すると、サーバーはレスポンスヘッダーでコンテンツタイプを text/event-stream と宣言し、接続を閉じずに連続してデータを送信します。クライアント側は、このデータストリームが終了するまで接続を維持し、サーバーから送られてくる新しいデータブロックを逐次処理します。
FastAPIでのSSE実装
PythonのモダンなWebフレームワークであるFastAPIは、非同期処理に優れており、SSEのようなストリーミング機能の実装に非常に適しています。FastAPIでSSEを扱うには、sse_starlette ライブラリを使用するのが一般的です。
環境構築
まず、必要なライブラリをインストールします。
pip install "fastapi[all]" sse-starlette uvicorn
サーバーサイドの実装
FastAPIアプリケーションでSSEエンドポイントを構築する際は、EventSourceResponse を利用します。以下の例では、定期的に現在時刻とカウンターをクライアントに送信するシンプルなSSEストリームを実装しています。
from fastapi import FastAPI, Request
from sse_starlette.sse import EventSourceResponse
import asyncio
import json
from datetime import datetime
# FastAPIアプリケーションの初期化
app = FastAPI()
# グローバルカウンター（イベントIDとして使用）
current_event_id = 0
@app.get("/event-stream") # SSEエンドポイントのパスを定義
async def get_event_stream(request: Request):
"""
クライアントにリアルタイムイベントをストリーミングするSSEエンドポイント。
"""
global current_event_id # グローバルカウンターにアクセス
async def generate_events():
nonlocal current_event_id # ネストされた関数で親スコープの変数を変更するため
while True:
# クライアントが切断されたかどうかを確認
if await request.is_disconnected():
print("クライアント接続が切断されました。ストリームを停止します。")
break
current_event_id += 1
timestamp_str = datetime.now().isoformat()
# クライアントへ送信するペイロードを生成
data_payload = {
"sequence_id": current_event_id,
"timestamp": timestamp_str,
"status": "processing",
"message": f"サーバーからのライブ更新データ #{current_event_id}"
}
# SSEフォーマットに準拠した辞書をyield
# sse_starletteがこれを自動的に "data: {...}\n\n" 形式に変換します
yield {
"event": "data_update", # イベントタイプ (クライアント側でフィルター可能)
"id": str(current_event_id), # イベントID
"retry": 5000, # クライアントが切断された場合に再接続を試みるまでの時間 (ミリ秒)
"data": json.dumps(data_payload) # データは文字列形式である必要があります
}
# 3秒ごとに新しいイベントを送信
await asyncio.sleep(3)
# EventSourceResponseを使ってストリーミングレスポンスを返す
return EventSourceResponse(generate_events())
このコードでは、/event-stream というパスでSSEエンドポイントを公開しています。generate_events 非同期ジェネレータ関数は無限ループで実行され、3秒ごとに新しいイベントデータを生成し、EventSourceResponse を通じてクライアントに送信します。クライアントが接続を切断した場合は、request.is_disconnected() が True を返し、ループを終了してリソースを解放します。
FastAPIアプリケーションを実行するには、以下のコマンドを使用します。
uvicorn main:app --reload
クライアントサイドのテストスクリプト
ブラウザのJavaScript (EventSource API) を使用してSSEストリームを購読するのが一般的ですが、ここではPythonの aiohttp ライブラリを使った簡単なテストクライアントを示します。これは、サーバーが正しくストリームを送信しているかを確認するのに役立ちます。
import aiohttp
import asyncio
async def receive_sse_events():
"""
FastAPIサーバーからSSEイベントを受信するクライアントスクリプト。
"""
server_url = "http://127.0.0.1:8000/event-stream" # サーバーのSSEエンドポイント
print(f"SSEストリームへ接続中: {server_url}...")
try:
async with aiohttp.ClientSession() as session:
# Acceptヘッダーでtext/event-streamを指定するのが推奨される
async with session.get(server_url, headers={'Accept': 'text/event-stream'}) as response:
response.raise_for_status() # HTTPエラーが発生した場合に例外を発生させる
print("接続確立。イベントを待機しています...")
# レスポンスのコンテンツをチャンクごとに読み込む
# 実際のSSEクライアントは "data:", "event:", "id:", "retry:" などの行をパースします
async for chunk in response.content.iter_any():
decoded_chunk = chunk.decode('utf-8').strip()
if decoded_chunk:
print(f"受信データ: {decoded_chunk}")
except aiohttp.ClientError as e:
print(f"クライアントエラーが発生しました: {e}")
except Exception as e:
print(f"予期せぬエラーが発生しました: {e}")
if __name__ == '__main__':
asyncio.run(receive_sse_events())
このPythonスクリプトは、指定されたURLに接続し、サーバーから送信されるデータチャンクをリアルタイムで受け取って表示します。実際のSSEイベントを適切に処理するには、受信したデータを行ごとに解析し、data:、event:、id:、retry: などのフィールドを分離する必要があります。
重要な注意事項
現在の多くのWebViewフレームワーク（特にモバイルアプリの組み込みWebViewなど）では、Server-Sent Events (SSE) サービスが意図した通りに動作しない、またはパフォーマンスの問題を抱える場合があります。クロスプラットフォームアプリケーションを開発する際は、WebSocketなど他のリアルタイム通信技術との互換性を検討することも重要です。