如何使用Server-Sent Events流式传输JSON数据？实现与疑问解答

基础文本流式传输示例 #

用FastAPI实现简单的文本流式传输，代码如下：

import time
from fastapi import FastAPI
from fastapi.responses import StreamingResponse

app = FastAPI()

def fake_data_streamer():
    for i in range(10):
        yield "some streamed data"
        time.sleep(0.5)

@app.get('/')
async def main():
    return StreamingResponse(fake_data_streamer())

客户端发起GET请求后，连接每0.5秒返回一次some streamed data。

你的JSON流式实现问题分析 #

你当前的JSON流式实现不符合SSE协议规范，会导致客户端无法正确识别事件边界，进而出现解析失败的情况：

def fake_data_streamer():
    for i in range(10):
        yield json.dumps({'result': 'a lot of streamed data', "seriously": ["so", "much", "data"]}, indent=4)
        time.sleep(0.5)

核心问题：

• 未遵循SSE事件格式要求，缺少data:前缀和\n\n事件分隔符
• 带缩进的JSON包含换行符，会被客户端SSE解析器误判为事件的一部分，导致多个JSON对象被拼接成无法解析的字符串

规范的服务端实现 #

要流式传输完整JSON对象，必须严格遵循SSE协议，每个事件封装为data: <完整JSON字符串>\n\n的格式，同时指定响应的MIME类型为text/event-stream：

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

app = FastAPI()

def fake_data_streamer():
    for i in range(10):
        # 生成完整JSON字符串
        json_payload = json.dumps({
            'result': 'a lot of streamed data', 
            "seriously": ["so", "much", "data"]
        })
        # 按SSE格式封装事件
        yield f"data: {json_payload}\n\n"
        time.sleep(0.5)

@app.get('/stream-json')
async def stream_json():
    return StreamingResponse(
        fake_data_streamer(), 
        media_type="text/event-stream"
    )

客户端正确读取方式 #

以浏览器端的EventSource为例，监听SSE的message事件，每次解析事件中的完整JSON：

const sse = new EventSource('/stream-json');

sse.onmessage = (event) => {
    try {
        const jsonData = JSON.parse(event.data);
        console.log('接收完整JSON:', jsonData);
    } catch (err) {
        console.error('JSON解析失败:', err);
    }
};

sse.onerror = (err) => {
    console.error('SSE连接错误:', err);
    sse.close();
};

参考OpenAI的实现逻辑 #

OpenAI的流式API同样基于SSE协议：

• 每个SSE事件的data字段对应一个完整的JSON对象（包含id、choices等核心字段）
• 若需增量返回内容（如逐字生成文本），则返回包含delta字段的JSON，由客户端自行拼接内容；若传输完整JSON对象，则每个事件对应一个可直接解析的完整JSON结构，与上述规范实现逻辑一致。

内容的提问来源于stack exchange，提问作者GenericDeveloperProfile