サーバー送信イベント (SSE) で API レスポンスをストリーミングする方法

要約

SSE（Server-Sent Events）を使用してHTTP経由でAPIレスポンスをストリーミングします。`Content-Type: text/event-stream` を送信し、イベントを `data: {json}\n\n` の形式で記述します。SSEは、AIレスポンスのストリーミング、進捗状況の更新、ライブフィードに機能します。Modern PetstoreAPIは、AIペット推薦と注文ステータスの更新にSSEを使用しています。

はじめに

あなたのAPIはペットのAI推薦を生成します。その応答には10秒かかります。ユーザーを待たせますか、それとも生成された結果をストリーミングしますか？

Server-Sent Events（SSE）を使用すると、リアルタイムでレスポンスをストリーミングできます。AIが結果を生成するにつれて、ユーザーはすぐにそれらを見ることができ、より良い体験を生み出します。

Modern PetstoreAPIは、AIペット推薦、注文ステータスの更新、在庫変更にSSEを使用しています。

ストリーミングAPIをテストする場合、ApidogはSSEのテストと検証をサポートしています。

ボタン

SSEの基本

SSEは、サーバーからクライアントへのHTTPベースの一方向ストリーミングです。

SSEのフォーマット

Content-Type: text/event-stream
Cache-Control: no-cache

data: {"message":"First chunk"}\n\n
data: {"message":"Second chunk"}\n\n
data: {"message":"Third chunk"}\n\n

各イベントは次のとおりです。

`data:` で始まります
ペイロードを含みます
`\n\n`（2つの改行）で終わります

名前付きイベント

event: recommendation
data: {"petId":"019b4132","score":0.95}

event: recommendation
data: {"petId":"019b4127","score":0.89}

event: complete
data: {"total":2}

イベントID

id: 1
data: {"message":"First"}

id: 2
data: {"message":"Second"}

クライアントは切断された場合、最後のIDから再開できます。

SSEサーバーの実装

Node.js/Expressの例

app.get('/v1/pets/recommendations/stream', async (req, res) => {
  // Set SSE headers
  res.setHeader('Content-Type', 'text/event-stream');
  res.setHeader('Cache-Control', 'no-cache');
  res.setHeader('Connection', 'keep-alive');

  // Send recommendations as they're generated
  const recommendations = await generateRecommendations(req.query.userId);

  for (const rec of recommendations) {
    res.write(`data: ${JSON.stringify(rec)}\n\n`);
    await sleep(100); // Simulate streaming delay
  }

  // Send completion event
  res.write(`event: complete\ndata: {"total":${recommendations.length}}\n\n`);
  res.end();
});

Python/FastAPIの例

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

app = FastAPI()

@app.get("/v1/pets/recommendations/stream")
async def stream_recommendations(user_id: str):
    async def generate():
        recommendations = await get_recommendations(user_id)

        for rec in recommendations:
            yield f"data: {json.dumps(rec)}\n\n"
            await asyncio.sleep(0.1)

        yield f"event: complete\ndata: {json.dumps({'total': len(recommendations)})}\n\n"

    return StreamingResponse(
        generate(),
        media_type="text/event-stream",
        headers={
            "Cache-Control": "no-cache",
            "Connection": "keep-alive"
        }
    )

SSEクライアントの実装

JavaScript/ブラウザ

const eventSource = new EventSource(
  'https://petstoreapi.com/v1/pets/recommendations/stream?userId=user-456'
);

eventSource.onmessage = (event) => {
  const data = JSON.parse(event.data);
  displayRecommendation(data);
};

eventSource.addEventListener('complete', (event) => {
  const data = JSON.parse(event.data);
  console.log(`Received ${data.total} recommendations`);
  eventSource.close();
});

eventSource.onerror = (error) => {
  console.error('SSE error:', error);
  eventSource.close();
};

Reactフック

import { useEffect, useState } from 'react';

function useSSE(url) {
  const [data, setData] = useState([]);
  const [complete, setComplete] = useState(false);

  useEffect(() => {
    const eventSource = new EventSource(url);

    eventSource.onmessage = (event) => {
      const item = JSON.parse(event.data);
      setData(prev => [...prev, item]);
    };

    eventSource.addEventListener('complete', () => {
      setComplete(true);
      eventSource.close();
    });

    return () => eventSource.close();
  }, [url]);

  return { data, complete };
}

// 使用例
function Recommendations({ userId }) {
  const { data, complete } = useSSE(
    `https://petstoreapi.com/v1/pets/recommendations/stream?userId=${userId}`
  );

  return (
    <div>
      {data.map(rec => (
        <PetCard key={rec.petId} pet={rec} />
      ))}
      {!complete && <Spinner />}
    </div>
  );
}

Modern PetstoreAPIがSSEをどのように利用しているか

AIペット推薦

AIが生成した推薦をストリーミングします。

GET /v1/pets/recommendations/stream?userId=user-456
Accept: text/event-stream

event: recommendation
data: {"petId":"019b4132","name":"Fluffy","score":0.95,"reason":"Matches your preference for cats"}

event: recommendation
data: {"petId":"019b4127","name":"Buddy","score":0.89,"reason":"Similar to pets you liked"}

event: complete
data: {"total":2,"processingTime":850}

注文ステータスの更新

注文処理のステップをストリーミングします。

GET /v1/orders/019b4132/status/stream
Accept: text/event-stream

data: {"status":"payment_processing","timestamp":"2026-03-13T10:30:00Z"}

data: {"status":"payment_confirmed","timestamp":"2026-03-13T10:30:02Z"}

data: {"status":"preparing_shipment","timestamp":"2026-03-13T10:30:05Z"}

event: complete
data: {"status":"shipped","trackingNumber":"1Z999AA10123456784"}

在庫の変更

リアルタイムの在庫更新をストリーミングします。

GET /v1/inventory/stream
Accept: text/event-stream

event: stock-change
data: {"petId":"019b4132","oldStock":5,"newStock":4}

event: price-change
data: {"petId":"019b4127","oldPrice":299.99,"newPrice":279.99}

Modern PetstoreAPI SSEドキュメントをご覧ください。

ApidogでのSSEテスト

ApidogはSSEテストをサポートしています。

SSEリクエストを作成する
`Accept: text/event-stream` を設定する
接続し、リアルタイムでイベントを表示する
イベント形式を検証する
再接続をテストする

ベストプラクティス

1. 適切なヘッダーを設定する

res.setHeader('Content-Type', 'text/event-stream');
res.setHeader('Cache-Control', 'no-cache');
res.setHeader('Connection', 'keep-alive');
res.setHeader('X-Accel-Buffering', 'no'); // Disable nginx buffering

2. ハートビートを送信する

接続を維持します。

const heartbeat = setInterval(() => {
  res.write(': heartbeat\n\n');
}, 15000);

res.on('close', () => clearInterval(heartbeat));

3. エラーを適切に処理する

eventSource.onerror = (error) => {
  if (eventSource.readyState === EventSource.CLOSED) {
    // Connection closed, will auto-reconnect
  } else {
    // Other error
    console.error('SSE error:', error);
  }
};

4. 再開のためにイベントIDを使用する

let lastEventId = 0;

app.get('/stream', (req, res) => {
  const startId = parseInt(req.headers['last-event-id'] || '0');

  for (let i = startId + 1; i <= 100; i++) {
    res.write(`id: ${i}\ndata: {"message":"Event ${i}"}\n\n`);
  }
});

5. 接続を閉じる

// クライアント
eventSource.addEventListener('complete', () => {
  eventSource.close();
});

// サーバー
res.on('close', () => {
  // リソースをクリーンアップ
});

まとめ

SSEはAPIレスポンスのストリーミングに最適です。一方向通信ではWebSocketよりもシンプルで、HTTP経由で動作し、自動的に再接続を処理します。

Modern PetstoreAPIは、AIストリーミング、注文更新、ライブフィードにSSEを使用しています。ApidogでSSEエンドポイントをテストしてください。

よくある質問

SSEは企業のファイアウォールを通過できますか？

はい、SSEは標準のHTTP/HTTPSを使用するため、ほとんどのファイアウォールやプロキシを通過できます。

SSE接続はどれくらい長く開いたままにできますか？

無期限ですが、プロキシを介して接続を維持するために15～30秒ごとにハートビートを使用してください。

SSE経由でバイナリデータを送信できますか？

いいえ、SSEはテキストのみです。バイナリデータはBase64エンコードするか、代わりにWebSocketを使用してください。

SSEは双方向通信をサポートしていますか？

いいえ、SSEはサーバーからクライアントへの一方向のみです。クライアントからサーバーへの通信には、通常のHTTPリクエストを使用します。

ブラウザはいくつのSSE接続を持つことができますか？

ブラウザはドメインあたりのSSE接続数を制限しています（通常6つ）。多数の接続には多重化またはWebSocketを使用してください。