You need to enable JavaScript to run this app.
优惠活动
大模型
产品
解决方案
定价
更多

Quart中Server Sent Events(SSE)无法正常流式输出的问题求助

Quart中实现Server Sent Events(SSE)端点的问题

尝试按照Quart官方文档实现SSE端点,复制示例代码并加入测试数据后,服务器因无限流式输出该测试数据直接锁死,说明官方示例无法直接运行,必须自行实现事件添加逻辑。

改用asyncio.queues搭建队列后,send_events()函数已能正常响应事件添加,但调用该端点时(通过Postman)始终没有输出,一直处于等待响应状态;只有中途停止服务器,才能获取到截至当时生成的SSE内容——这说明事件已触发且格式正确,但输出没有按预期进行流式传输。

曾找到过类似问题的讨论,但最终方案是针对Starlette框架实现的。尝试使用相关包(同时我也在使用FastAPI),但在SSE端点返回EventSourceResponse时出现错误:

TypeError: The response value type (EventSourceResponse) is not valid

显然Quart不支持该响应类型,而官方示例又无法正常运行,难道只能放弃Quart?有没有其他可行的解决方案?

代码

数据类定义

@dataclass
class ServerSentEvent:
    data: str
    event: str

    def encode(self) -> bytes:
        message = f'data: {self.data}'
        message = f'{message}\nevent: {self.event}'
        message = f'{message}\r\n\r\n'
        return message.encode('utf-8')

SSE端点实现

queues = []

@bp.get('/sse')
async def sse():
    """Server Sent Events"""

    if 'text/event-stream' not in request.accept_mimetypes:
        abort(400)

    async def send_events():
        while True:
            print('waiting for event')
            event = await queue.get()
            print('got event')
            print(event.encode())
            yield event.encode()

    queue = Queue()
    queues.append(queue)

    await queue.put(ServerSentEvent(**{'event': 'subscribed', 'data': 'will send new events'}))
    print('-- SSE connected')

    response = await make_response(
        send_events(),
        {
            'Content-Type': 'text/event-stream',
            'Cache-Control': 'no-cache',
            'Transfer-Encoding': 'chunked',
            'Connection': 'keep-alive'
        }
    )
    response.timeout = None
    return response

内容的提问来源于stack exchange,提问作者kasimir

相关产品推荐
方舟 Agent Plan

超全模态模型 × Harness 升级,最新支持 Deepseek-V4.1-Flash、GLM-5.3 系列、Doubao-Seedream-5.0-pro、Kimi-K3 (部分), 限时 9.9 元起

最近更新时间:2026.06.27 20:13:21