通过HttpServerResponse SSE能力结合扣子（CozeAPI）无法流式返回

[miser]

Describe the problem（描述问题）

Hi，
我在扣子上搭建了一个工作流，我想通过midway调用它，并以SSE返回数据，但是我发现，按照CozeAPI和midway的例子去实现功能，总是用等整个Coze API所有数据请求结束后才会将所有数据一起返回，打个比方 CozeAPI返回 [1,2,3,4,5]，按照SSE来说，会返回1 -> 2 -> 3 -> 4 -> 5。但是实践后发现是 1,2,3,4,5 一起返回，没有流式的效果。但是同样的CozeAPI代码，我通过express实现是可以实现类似效果的。

请问这是什么原因？

Midway代码

 @Post('/qStream')
  async queryTrendStream(@Body() { question }: { question: string }) {
    const apiClient = new CozeAPI({
      token: 'xxxxx',
      baseURL: 'https://api.coze.cn',
    });

    const workflow = await apiClient.workflows.chat.stream({
      workflow_id: '12345',
      additional_messages: [
        {
          content: question,
          content_type: 'text',
          role: RoleType.User,
          type: 'question',
        },
      ],
      app_id: 'xxxx',
    });

    const res = new HttpServerResponse(this.ctx).sse();
   

    for await (const event of workflow) {
      res.sendEnd({
        data: event,
      });
    }
    res.end();
    return res;
  }

express代码

import express from 'express';
import cors from 'cors';
import helmet from 'helmet';
import morgan from 'morgan';
import compression from 'compression';
import SSE from 'express-sse';

import { CozeAPI, RoleType } from '@coze/api';

const app = express();
const port = process.env.PORT || 7001;
const sse = new SSE();

app.post('/api/target/qStream', async (req, res) => {
  sse.init(req, res);
  const apiClient = new CozeAPI({
    token: 'xxxxxx',
    baseURL: 'https://api.coze.cn',
  });

  const workflow = apiClient.workflows.chat.stream({
    workflow_id: 'xxxx',
    additional_messages: [
      {
        content: 'xxxx',
        content_type: 'text',
        role: RoleType.User,
        type: 'question',
      },
    ],
    app_id: 'xxxx',
  });

  for await (const event of workflow) {
    sse.send({ data: event }, 'update');
  }

  sse.sendEnd();
});

Midway Versions（Midway 版本）

@midwayjs/faas-typings(not installed)
✓ @midwayjs/fc-starter(not installed)
✓ @midwayjs/serverless-http-parser(not installed)
✓ @midwayjs/async-hooks-context-manager(not installed)
✓ @midwayjs/axios(not installed)
✓ @midwayjs/bootstrap(3.20.0)
✓ @midwayjs/bull(not installed)
✓ @midwayjs/bull-board(not installed)
✓ @midwayjs/bullmq(not installed)
✓ @midwayjs/busboy(not installed)
✓ @midwayjs/cache-manager(not installed)
✓ @midwayjs/captcha(not installed)
✓ @midwayjs/casbin(not installed)
✓ @midwayjs/casbin-redis-adapter(not installed)
✓ @midwayjs/casbin-typeorm-adapter(not installed)
✓ @midwayjs/code-dye(not installed)
✓ @midwayjs/consul(not installed)
✓ @midwayjs/core(3.20.0)
✓ @midwayjs/cos(not installed)
✓ @midwayjs/cron(3.20.0)
✓ @midwayjs/cross-domain(3.20.0)
✓ @midwayjs/decorator(not installed)
✓ @midwayjs/etcd(not installed)
✓ @midwayjs/express-session(not installed)
✓ @midwayjs/faas(not installed)
✓ @midwayjs/grpc(not installed)
✓ @midwayjs/http-proxy(not installed)
✓ @midwayjs/i18n(not installed)
✓ @midwayjs/info(not installed)
✓ @midwayjs/jwt(not installed)
✓ @midwayjs/kafka(not installed)
✓ @midwayjs/leoric(not installed)
✓ @midwayjs/mikro(not installed)
✓ @midwayjs/mock(3.20.0)
✓ @midwayjs/mongoose(3.20.1)
✓ @midwayjs/mqtt(not installed)
✓ @midwayjs/oss(not installed)
✓ @midwayjs/otel(not installed)
✓ @midwayjs/passport(not installed)
✓ @midwayjs/process-agent(not installed)
✓ @midwayjs/prometheus(not installed)
✓ @midwayjs/prometheus-socket-io(not installed)
✓ @midwayjs/rabbitmq(not installed)
✓ @midwayjs/redis(not installed)
✓ @midwayjs/security(not installed)
✓ @midwayjs/sequelize(not installed)
✓ @midwayjs/session(not installed)
✓ @midwayjs/socketio(not installed)
✓ @midwayjs/static-file(not installed)
✓ @midwayjs/swagger(not installed)
✓ @midwayjs/tablestore(not installed)
✓ @midwayjs/tags(not installed)
✓ @midwayjs/tenant(not installed)
✓ @midwayjs/typegoose(3.20.1)
✓ @midwayjs/typeorm(not installed)
✓ @midwayjs/upload(not installed)
✓ @midwayjs/validate(not installed)
✓ @midwayjs/version(not installed)
✓ @midwayjs/view(not installed)
✓ @midwayjs/view-ejs(not installed)
✓ @midwayjs/view-nunjucks(not installed)
✓ @midwayjs/web(3.20.0)
✓ @midwayjs/express(not installed)
✓ @midwayjs/koa(not installed)
✓ @midwayjs/ws(not installed)

[czy88840616]

是 send 吧?

[miser]

是的 ，我的例子里面写错了，res.send还是一样的。

midwayjs

Jietu20250612-140334.mp4

express

Jietu20250612-140848.mp4

[miser]

我自己实现一个简单的sse方法 就可以了

@Post('/qStream')
  async queryTrendStream(@Body() { question }: { question: string }) {
    this.ctx.set({
      'Content-Type': 'text/event-stream',
      'Cache-Control': 'no-cache',
      Connection: 'keep-alive',
      'Access-Control-Allow-Origin': '*',
    });

    const send = (data: any, event?: string) => {
      if (event) {
        this.ctx.res.write(`event: ${event}\n`);
      }
      this.ctx.res.write(`data: ${JSON.stringify(data)}\n\n`);
    };

    const end = () => {
      this.ctx.res.end();
    };

    this.ctx.sse = { send, end };

    const apiClient = new CozeAPI({
      token:
        'xxx',
      baseURL: 'https://api.coze.cn',
    });

    const workflow = await apiClient.workflows.chat.stream({
      workflow_id: 'xx',
      additional_messages: [
        {
          content: question,
          content_type: 'text',
          role: RoleType.User,
          type: 'question',
        },
      ],
      app_id: 'xx',
    });

    for await (const event of workflow) {
      this.ctx.sse.send({
        data: event,
      });
    }
    this.ctx.sse.end();
    return {};
  }

[czy88840616]

额，我得搭一个 coze 试试。

[ghostoy]

前面for await要执行完了才return res，没有流式的效果。所以，必须先返回res再执行异步方法。

有几个办法：

1. 用setImmediate或者nextTick，让中间的所有异步过程在返回后执行
2. 将中间的异步过程包装成另一个async函数，使用.then().catch()的方式调用执行

[czy88840616]

补充一下这里的关键点，@ghostoy 的判断是对的：如果在 controller 方法里先 for await 把上游流消费完，再 return res，框架层拿到响应对象时上游已经结束了，客户端自然只能一次性收到结果。

正确模式是先把 SSE response 返回给框架，然后在异步任务里写入数据，例如：

@Post("/qStream")
async queryTrendStream(@Body() { question }: { question: string }) {
  const res = new HttpServerResponse(this.ctx).sse();

  setImmediate(async () => {
    try {
      const workflow = await apiClient.workflows.chat.stream(/* ... */);
      for await (const event of workflow) {
        res.send({ data: event });
      }
      res.sendEnd({ data: "end" });
    } catch (err) {
      res.sendError(err);
    }
  });

  return res;
}

如果你已经用原生 ctx.res.write 跑通，也可以继续用那个方式。新版本里 HttpServerResponse.sse() 也增加了更直接的 forward() 能力，适合把 SDK 的 AsyncIterable 流直接转成 SSE；v3 项目如果不升级，就按上面“先 return，再异步写”的方式处理。