---
url: https://vafast.dev/essential/sse.md
---
# Server-Sent Events (SSE)

Vafast 提供了内置的 SSE 支持，用于实现流式响应，如 AI 聊天、实时进度更新、通知推送等场景。

## 框架集成方式

声明 `sse: true` 后，无需手写 `ReadableStream`。`defineRoute` 在扁平化时会自动：

1. 用 `wrapGeneratorToSSEHandler` 把你的 `async function*` 转为 SSE 响应
2. 用 `wrapSSEHandler` 走与普通路由相同的 schema 验证、中间件上下文（`userInfo` 等）

```typescript
// sse: true + async function*
defineRoute({
  sse: true,
  handler: async function* (ctx) {
    yield { type: 'start' }
    yield { type: 'done' }
  },
})
```

handler 与普通路由一样享有完整上下文：`params` / `query` / `body` / `headers`，以及中间件 `next({ userInfo })` 注入的字段。

## 快速开始

通过 `sse: true` 显式声明 SSE 端点，handler 使用 `async function*` 语法。直接 `yield` 任意数据，框架自动包装为 SSE 格式：

```typescript
import { defineRoute, defineRoutes, Type, sse } from 'vafast'

const routes = defineRoutes([
  defineRoute({
    method: 'GET',
    path: '/progress',
    sse: true,
    handler: async function* () {
      // 直接 yield 数据，框架自动包装为 SSE data 字段
      yield { status: 'started' }
      
      for (let i = 0; i <= 100; i += 10) {
        yield { progress: i }
        await new Promise(r => setTimeout(r, 100))
      }
      
      // 需要自定义事件名时，使用 sse() 函数
      yield sse({ event: 'complete' }, { message: 'Done!' })
    },
  }),
])
```

## 两种使用模式

### 简单模式（推荐）

直接 `yield` 任意数据，框架自动序列化为 SSE `data` 字段：

```typescript
// yield 对象
yield { type: 'text_delta', content: 'Hello' }
// 输出: data: {"type":"text_delta","content":"Hello"}

// yield 字符串
yield 'Hello World'
// 输出: data: Hello World

// yield 数字
yield 42
// 输出: data: 42
```

::: tip 为什么推荐简单模式？

* **代码更简洁** — 无需包装每个事件
* **符合直觉** — `yield data` 直接发送数据
* **类型友好** — 与 AI SDK 的 `{ type, data }` ChatEvent 等类型无缝集成
  :::

::: warning 不要混用旧格式
`yield { event: 'done', data: {...} }` **不会**自动变成 SSE 的 `event:` 行，整段对象会作为 `data` 字段发出。需要自定义事件名请用 `sse({ event: 'done' }, payload)`。
:::

### 高级模式

需要设置 SSE 的 `event`、`id`、`retry` 元数据时，使用 `sse()` 函数：

```typescript
import { sse } from 'vafast'

// 自定义事件名称
yield sse({ event: 'status' }, { online: true })
// 输出: event: status
//       data: {"online":true}

// 带 ID（支持断线重连）
yield sse({ id: '1001' }, { value: 1 })
// 输出: id: 1001
//       data: {"value":1}

// 自定义重试间隔
yield sse({ retry: 5000 }, 'reconnect')
// 输出: retry: 5000
//       data: reconnect

// 完整配置
yield sse({ event: 'update', id: '42', retry: 3000 }, { count: 1 })
// 输出: id: 42
//       event: update
//       retry: 3000
//       data: {"count":1}
```

## 基础用法

### GET + Query 参数

适用于简单的订阅场景：

```typescript
defineRoute({
  method: 'GET',
  path: '/tasks/:id/stream',
  sse: true,
  schema: {
    params: Type.Object({ id: Type.String() }),
  },
  handler: async function* ({ params }) {
    yield { taskId: params.id, status: 'streaming' }
    // ... 业务逻辑
  },
})
```

### POST + Body（AI 场景）

适用于需要发送复杂数据的场景（如 AI 聊天）：

```typescript
defineRoute({
  method: 'POST',
  path: '/chat/stream',
  sse: true,
  schema: {
    body: Type.Object({
      messages: Type.Array(Type.Object({
        role: Type.String(),
        content: Type.String(),
      })),
      model: Type.Optional(Type.String()),
    }),
  },
  handler: async function* ({ body }) {
    const { messages, model = 'gpt-4' } = body
    
    // 简单模式：直接 yield AI 事件
    yield { type: 'start', model }
    
    for await (const chunk of aiStream(messages)) {
      yield { type: 'text_delta', content: chunk.text }
    }
    
    yield { type: 'done', usage: { tokens: 100 } }
  },
})
```

## 与路由组组合

先定义 SSE handler 常量，再挂到 `defineRoutes`（与普通路由相同）：

```typescript
import { defineRoute, defineRoutes, Type } from 'vafast'

const progressHandler = defineRoute({
  method: 'GET',
  path: '/:id/progress',
  name: '订阅任务进度',
  sse: true,
  schema: {
    params: Type.Object({ id: Type.String() }),
  },
  handler: async function* ({ params }) {
    const task = await getTask(params.id)
    if (!task) {
      yield { error: '记录不存在' }
      return
    }

    yield { status: task.status, progress: task.progress }

    while (task.status === 'running') {
      await sleep(2000)
      const updated = await getTask(params.id)
      if (!updated) return
      yield { status: updated.status, progress: updated.progress }
      if (updated.status === 'success' || updated.status === 'failed') return
    }
  },
})

export const taskRoutes = defineRoutes([
  defineRoute({
    path: '/api/tasks',
    children: [
      progressHandler,
      // 其它 REST handler...
    ],
  }),
])
```

需要登录态时，用教程同款的 `defineMiddleware`，或生产场景的 [@vafast/auth-middleware](/middleware/auth-middleware) 包装路由（把上面的 `defineRoute` 换成对应的 `defineAuthRoute` 即可）。

AI 流式对话：`yield` 事件对象，也可直接转发业务生成器：

```typescript
const streamHandler = defineRoute({
  method: 'POST',
  path: '/stream',
  sse: true,
  schema: { body: Type.Object({ message: Type.String() }) },
  handler: async function* ({ body }) {
    for await (const event of streamAgentRun(body.message)) {
      yield event // { type: 'text_delta', data: { ... } }
    }
  },
})
```

## 错误处理

### Generator 内部错误

如果 generator 函数抛出错误，会自动发送一个 `error` 事件：

```typescript
defineRoute({
  method: 'GET',
  path: '/stream',
  sse: true,
  handler: async function* () {
    yield { status: 'processing' }
    throw new Error('Something went wrong')
    // 客户端会收到: event: error
    //              data: {"error":"Something went wrong"}
  },
})
```

### Schema 验证错误

如果 schema 验证失败，会返回 **422** 错误（JSON 格式）：

```typescript
defineRoute({
  method: 'GET',
  path: '/stream',
  sse: true,
  schema: { query: Type.Object({ required: Type.String() }) },
  handler: async function* () { /* ... */ },
})

// 请求 /stream（缺少 required 参数）
// HTTP 422
// { "code": 422, "message": "请求参数校验失败", "details": [...] }
```

## 实际应用场景

### 1. 视频/文件处理进度

```typescript
defineRoute({
  method: 'GET',
  path: '/tasks/:taskId/progress',
  sse: true,
  schema: {
    params: Type.Object({ taskId: Type.String() }),
  },
  handler: async function* ({ params }) {
    const { taskId } = params
    
    while (true) {
      const task = await getTask(taskId)
      
      yield { 
        status: task.status,
        progress: task.progress,
      }
      
      if (task.status === 'completed' || task.status === 'failed') {
        return
      }
      
      await new Promise(r => setTimeout(r, 2000))
    }
  },
})
```

### 2. AI 聊天流式响应

```typescript
defineRoute({
  method: 'POST',
  path: '/chat/stream',
  sse: true,
  schema: {
    body: Type.Object({
      messages: Type.Array(Type.Object({
        role: Type.Union([Type.Literal('user'), Type.Literal('assistant')]),
        content: Type.String(),
      })),
    }),
  },
  handler: async function* ({ body }) {
    const { messages } = body
    
    // 使用简单模式，直接 yield ChatEvent 格式
    yield { type: 'start', timestamp: Date.now() }
    
    for await (const chunk of aiStream(messages)) {
      yield { type: 'text_delta', content: chunk.text }
    }
    
    yield { 
      type: 'done', 
      usage: { promptTokens: 100, completionTokens: 50 },
    }
  },
})
```

### 3. 实时通知推送

```typescript
import { sse } from 'vafast'

defineRoute({
  method: 'GET',
  path: '/notifications/stream',
  sse: true,
  schema: {
    query: Type.Object({ userId: Type.String() }),
  },
  handler: async function* ({ query }) {
    const { userId } = query
    
    for await (const notification of subscribeNotifications(userId)) {
      // 使用 sse() 设置事件名和 ID
      yield sse(
        { event: notification.type, id: notification.id },
        notification.payload
      )
    }
  },
})
```

## 响应头

SSE 端点会自动设置以下响应头：

| Header | Value | 说明 |
|--------|-------|------|
| `Content-Type` | `text/event-stream` | SSE MIME 类型 |
| `Cache-Control` | `no-cache` | 禁用缓存 |
| `Connection` | `keep-alive` | 保持连接 |
| `X-Accel-Buffering` | `no` | Nginx 禁用缓冲 |

## 客户端使用

### @vafast/api-client（推荐）

类型安全客户端通过链式 `.sse()` 消费流式接口，无需手写解析：

```typescript
import { createApiClient } from './api.generated'

const api = createApiClient(client)

// POST SSE — AI 对话
api.agent.stream.post({ message: '你好' }).sse({
  onMessage: (data) => {
    // data 即服务端 yield 的 JSON（如 { type, data }）
    if (data.type === 'text_delta') process.stdout.write(data.data?.content ?? '')
  },
  onError: (error) => console.error(error),
})

// GET SSE — 进度订阅
api.videoGeneration.progress.get({ id: taskId }).sse({
  onMessage: (data) => console.log(data.progress),
})
```

> 📖 详见 [API Client 概览](/api-client/overview#sse-流式响应)

### 浏览器原生 EventSource（GET 请求）

```javascript
const eventSource = new EventSource('/api/progress/123')

eventSource.onmessage = (event) => {
  const data = JSON.parse(event.data)
  console.log('Progress:', data.progress)
}

eventSource.addEventListener('complete', (event) => {
  console.log('Done!')
  eventSource.close()
})

eventSource.onerror = (error) => {
  console.error('Error:', error)
}
```

::: warning 注意
`EventSource` **不支持自定义请求头**（如 Authorization）和 **POST 请求**。如果需要认证或发送请求体，请使用 `fetch + ReadableStream`。
:::

### fetch + ReadableStream（支持 POST 和认证）

```javascript
async function subscribeSSE(url, body, token) {
  const response = await fetch(url, {
    method: 'POST',
    headers: {
      'Content-Type': 'application/json',
      'Authorization': `Bearer ${token}`,
    },
    body: JSON.stringify(body),
  })
  
  const reader = response.body.getReader()
  const decoder = new TextDecoder()
  
  while (true) {
    const { done, value } = await reader.read()
    if (done) break
    
    const text = decoder.decode(value)
    const lines = text.split('\n')
    for (const line of lines) {
      if (line.startsWith('data: ')) {
        const data = JSON.parse(line.slice(6))
        console.log('Received:', data)
      }
    }
  }
}
```

## 最佳实践

::: tip 建议

1. **直接 yield 数据** — 简单模式覆盖 90% 场景，代码更简洁
2. **需要 event/id/retry 时用 `sse()`** — 高级模式提供完整控制
3. **定期发送心跳** — 防止连接被中间代理断开
4. **使用事件 ID** — 支持断线重连时从上次位置继续
5. **设置合理的重试间隔** — 避免客户端频繁重连
6. **优雅处理错误** — 框架自动发送 error 事件
7. **资源清理** — 在 generator 结束时清理定时器、数据库连接等
   :::