> ## Documentation Index
> Fetch the complete documentation index at: https://docs.shannon.run/llms.txt
> Use this file to discover all available pages before exploring further.

# 流式端点

> 通过 SSE 和 WebSocket 实时事件流传输

## 概述

Shannon 通过 Server-Sent Events (SSE) 和 WebSocket 协议提供实时事件流传输。使用流式传输来监控任务执行、显示进度并在生成时接收结果。

<Warning>
  **认证**：流式端点与其他 API 使用相同的认证请求头。
  浏览器 EventSource 无法携带自定义请求头。

  * 开发环境：设置 `GATEWAY_SKIP_AUTH=1`。
  * 生产环境：通过后端代理转发 SSE，并注入 `X-API-Key` 或 Bearer 头。
    SSE 端点支持 `api_key` 查询参数作为回退（如 `?api_key=sk_...`）。其他端点请通过请求头传递密钥。
</Warning>

<Note>
  **流式传输限制**：

  * **超时**：流在 5 分钟无活动后自动关闭
  * **缓冲区大小**：每个连接最大 1MB 缓冲数据
  * **使用元数据**：现在所有 LLM 提供商（OpenAI、Anthropic、Google、Groq、xAI）都可获取 token 计数和成本
</Note>

## 端点

| 方法     | 端点                          | 协议        | 描述                    |
| ------ | --------------------------- | --------- | --------------------- |
| `POST` | `/api/v1/tasks/stream`      | HTTP+SSE  | 提交任务并获取流 URL（推荐）      |
| `GET`  | `/api/v1/stream/sse`        | SSE       | Server-Sent Events 端点 |
| `GET`  | `/api/v1/stream/ws`         | WebSocket | WebSocket 流式传输端点      |
| `GET`  | `/api/v1/tasks/{id}/events` | HTTP      | 获取历史事件（分页）            |

## 统一提交 + 流式传输（推荐）

### POST /api/v1/tasks/stream

提交任务并立即开始流式传输其事件的最简单方法。此端点在一次调用中结合了任务提交和流式设置。

<Tip>
  **最适合前端应用**：此端点非常适合需要在提交任务后立即显示进度的实时 UI。
</Tip>

#### 认证

**必需**：是

```
X-API-Key: sk_test_123456
```

或：

```
Authorization: Bearer YOUR_TOKEN
```

#### 请求体

| 参数                  | 类型     | 必需 | 描述                                       |
| ------------------- | ------ | -- | ---------------------------------------- |
| `query`             | string | 是  | 自然语言任务描述                                 |
| `session_id`        | string | 否  | 多轮对话的会话标识符                               |
| `context`           | object | 否  | 键值对形式的附加上下文数据                            |
| `model_tier`        | string | 否  | 模型层级：`small`、`medium`、`large`            |
| `model_override`    | string | 否  | 指定模型名称（规范 ID；例如 `gpt-5`）                 |
| `provider_override` | string | 否  | 强制指定提供商（如 `openai`、`anthropic`、`google`） |

#### 响应

**状态**：`201 Created`

**响应体**：

```json theme={"dark"}
{
  "task_id": "task_01HQZX3Y9K8M2P4N5S7T9W2V",
  "workflow_id": "task_01HQZX3Y9K8M2P4N5S7T9W2V",
  "stream_url": "/api/v1/stream/sse?workflow_id=task_01HQZX3Y9K8M2P4N5S7T9W2V"
}
```

#### 响应字段

| 字段            | 类型     | 描述             |
| ------------- | ------ | -------------- |
| `task_id`     | string | 唯一任务标识符        |
| `workflow_id` | string | 任务/工作流标识符      |
| `stream_url`  | string | SSE 流端点的相对 URL |

#### 示例：JavaScript/TypeScript

```javascript theme={"dark"}
async function submitAndStream(query, onEvent, onComplete, onError) {
  try {
    // 1. 提交任务并获取流 URL
    const response = await fetch('http://localhost:8080/api/v1/tasks/stream', {
      method: 'POST',
      headers: {
        'Content-Type': 'application/json',
        'Authorization': 'Bearer YOUR_TOKEN'
      },
      body: JSON.stringify({
        query: query,
        session_id: `session-${Date.now()}`
      })
    });

    if (!response.ok) {
      throw new Error(`HTTP ${response.status}: ${response.statusText}`);
    }

    const { task_id, workflow_id, stream_url } = await response.json();
    console.log(`任务已提交: ${workflow_id}`);

    // 2. 连接到 SSE 流
    // 注意：浏览器 EventSource 不支持自定义请求头。
    // 使用 api_key 查询参数，或通过后端代理转发 SSE。
    const eventSource = new EventSource(
      `http://localhost:8080${stream_url}&api_key=${localStorage.getItem('token')}`
    );

    eventSource.onmessage = (e) => {
      const event = JSON.parse(e.data);
      onEvent(event);

      // 检查完成状态
      if (event.type === 'WORKFLOW_COMPLETED') {
        eventSource.close();
        onComplete(event);
      }
    };

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

    return { workflow_id, eventSource };

  } catch (error) {
    onError(error);
    throw error;
  }
}

// 使用示例
submitAndStream(
  "分析第四季度收入趋势",
  (event) => console.log(`[${event.type}]`, event.message),
  (final) => console.log("已完成:", final.result),
  (error) => console.error("错误:", error)
);
```

#### 示例：React Hook

```jsx theme={"dark"}
import { useState, useCallback, useRef } from 'react';

function useTaskStream(apiUrl = 'http://localhost:8080') {
  const [events, setEvents] = useState([]);
  const [isStreaming, setIsStreaming] = useState(false);
  const [error, setError] = useState(null);
  const [workflowId, setWorkflowId] = useState(null);
  const eventSourceRef = useRef(null);

  const submitTask = useCallback(async (query, sessionId = null) => {
    setEvents([]);
    setError(null);
    setIsStreaming(true);

    try {
      // 提交任务
      const response = await fetch(`${apiUrl}/api/v1/tasks/stream`, {
        method: 'POST',
        headers: {
          'Content-Type': 'application/json',
          'Authorization': `Bearer ${localStorage.getItem('token')}`
        },
        body: JSON.stringify({
          query,
          session_id: sessionId || `session-${Date.now()}`
        })
      });

      if (!response.ok) throw new Error(`HTTP ${response.status}`);

      const { task_id, workflow_id, stream_url } = await response.json();
      setWorkflowId(workflow_id);

      // 连接到流
      // 注意：浏览器 EventSource 不支持自定义请求头，使用 api_key 查询参数
      const token = localStorage.getItem('token');
      const sseUrl = token
        ? `${apiUrl}${stream_url}&api_key=${token}`
        : `${apiUrl}${stream_url}`;
      const eventSource = new EventSource(sseUrl);
      eventSourceRef.current = eventSource;

      eventSource.onmessage = (e) => {
        const event = JSON.parse(e.data);
        setEvents(prev => [...prev, event]);

        if (event.type === 'WORKFLOW_COMPLETED' || event.type === 'ERROR_OCCURRED') {
          eventSource.close();
          setIsStreaming(false);
        }
      };

      eventSource.onerror = (err) => {
        setError(err);
        setIsStreaming(false);
        eventSource.close();
      };

    } catch (err) {
      setError(err);
      setIsStreaming(false);
    }
  }, [apiUrl]);

  const stopStreaming = useCallback(() => {
    if (eventSourceRef.current) {
      eventSourceRef.current.close();
      setIsStreaming(false);
    }
  }, []);

  return { submitTask, stopStreaming, events, isStreaming, error, workflowId };
}

// 在组件中使用
function TaskStreamDemo() {
  const { submitTask, stopStreaming, events, isStreaming, error, workflowId } = useTaskStream();

  return (
    <div>
      <button
        onClick={() => submitTask("15 + 25 等于多少？")}
        disabled={isStreaming}
      >
        {isStreaming ? '处理中...' : '提交任务'}
      </button>

      {workflowId && <p>工作流 ID: {workflowId}</p>}
      {error && <p style={{color: 'red'}}>错误: {error.message}</p>}

      <div>
        {events.map((event, idx) => (
          <div key={idx}>
            <strong>{event.type}</strong>: {event.message || event.agent_id}
          </div>
        ))}
      </div>

      {isStreaming && <button onClick={stopStreaming}>停止</button>}
    </div>
  );
}
```

#### 示例：Vue 3 Composition API

```vue theme={"dark"}
<script setup>
import { ref } from 'vue';

const events = ref([]);
const isStreaming = ref(false);
const error = ref(null);
const workflowId = ref(null);
let eventSource = null;

async function submitTask(query) {
  events.value = [];
  error.value = null;
  isStreaming.value = true;

  try {
    const response = await fetch('http://localhost:8080/api/v1/tasks/stream', {
      method: 'POST',
      headers: {
        'Content-Type': 'application/json',
        'Authorization': `Bearer ${localStorage.getItem('token')}`
      },
      body: JSON.stringify({
        query,
        session_id: `session-${Date.now()}`
      })
    });

    if (!response.ok) throw new Error(`HTTP ${response.status}`);

    const { workflow_id, stream_url } = await response.json();
    workflowId.value = workflow_id;

    // 浏览器 EventSource 不支持自定义请求头，使用 api_key 查询参数
    eventSource = new EventSource(`http://localhost:8080${stream_url}&api_key=${localStorage.getItem('token')}`);

    eventSource.onmessage = (e) => {
      const event = JSON.parse(e.data);
      events.value.push(event);

      if (event.type === 'WORKFLOW_COMPLETED') {
        eventSource.close();
        isStreaming.value = false;
      }
    };

    eventSource.onerror = (err) => {
      error.value = err;
      eventSource.close();
      isStreaming.value = false;
    };
  } catch (err) {
    error.value = err;
    isStreaming.value = false;
  }
}

function stopStreaming() {
  if (eventSource) {
    eventSource.close();
    isStreaming.value = false;
  }
}
</script>

<template>
  <div>
    <button @click="submitTask('分析数据')" :disabled="isStreaming">
      {{ isStreaming ? '处理中...' : '提交任务' }}
    </button>

    <div v-if="workflowId">工作流: {{ workflowId }}</div>
    <div v-if="error" style="color: red">错误: {{ error.message }}</div>

    <div v-for="(event, idx) in events" :key="idx">
      <strong>{{ event.type }}</strong>: {{ event.message || event.agent_id }}
    </div>

    <button v-if="isStreaming" @click="stopStreaming">停止</button>
  </div>
</template>
```

#### 示例：Python

```python theme={"dark"}
import httpx
import json
import time

def submit_and_stream(query: str, api_key: str):
    """提交任务并流式传输事件。"""

    # 1. 提交任务
    response = httpx.post(
        "http://localhost:8080/api/v1/tasks/stream",
        headers={
            "Authorization": f"Bearer {api_key}",
            "Content-Type": "application/json"
        },
        json={
            "query": query,
            "session_id": f"session-{int(time.time())}"
        }
    )

    data = response.json()
    workflow_id = data["workflow_id"]
    stream_url = data["stream_url"]

    print(f"任务已提交: {workflow_id}")

    # 2. 流式传输事件
    with httpx.stream(
        "GET",
        f"http://localhost:8080{stream_url}",
        headers={"Authorization": f"Bearer {api_key}"},
        timeout=None
    ) as stream_response:
        for line in stream_response.iter_lines():
            if line.startswith("data:"):
                event = json.loads(line[5:])
                print(f"[{event['type']}] {event.get('message', '')}")

                if event['type'] in ['WORKFLOW_COMPLETED']:
                    return event

# 使用示例
result = submit_and_stream("法国的首都是什么？", "sk_test_123456")
print("最终结果:", result.get('result'))
```

<Note>
  **为什么使用此端点？** 统一端点确保您在提交后立即开始流式传输，防止在分别提交和连接时可能错过的任何事件。
</Note>

## Server-Sent Events (SSE)

### GET /api/v1/stream/sse

使用 Server-Sent Events 进行实时事件流传输。

#### 认证

**必需**：是

```
X-API-Key: sk_test_123456
```

#### 查询参数

| 参数              | 类型     | 必需 | 描述                                                                                                  |
| --------------- | ------ | -- | --------------------------------------------------------------------------------------------------- |
| `workflow_id`   | string | 是  | 任务/工作流标识符                                                                                           |
| `types`         | string | 否  | 逗号分隔的事件类型以筛选                                                                                        |
| `last_event_id` | string | 否  | 从特定事件 ID 恢复。支持 Redis Stream ID（如 `1700000000000-0`）或数字序号（如 `42`）。当为数字时，重放规则为 `seq > last_event_id`。 |

#### 事件格式

每个事件遵循 SSE 规范：

```
id: <event_id>
event: <event_type>
data: <json_payload>

```

#### 示例请求

```bash theme={"dark"}
curl -N "http://localhost:8080/api/v1/stream/sse?workflow_id=task_abc123" \\
  -H "X-API-Key: sk_test_123456"
```

#### 示例响应

```
id: 1719000000000-0
event: WORKFLOW_STARTED
data: {"workflow_id":"task_abc123","type":"WORKFLOW_STARTED","agent_id":"orchestrator","message":"Starting up","timestamp":"2025-10-22T10:30:00Z","seq":1,"stream_id":"1719000000000-0"}

id: 1719000001000-0
event: AGENT_THINKING
data: {"workflow_id":"task_abc123","type":"AGENT_THINKING","agent_id":"Ryogoku","message":"Analyzing query...","timestamp":"2025-10-22T10:30:01Z","seq":2,"stream_id":"1719000001000-0"}

id: 1719000002000-0
event: TOOL_INVOKED
data: {"workflow_id":"task_abc123","type":"TOOL_INVOKED","agent_id":"Ryogoku","message":"Calling web_search","payload":{"tool":"web_search","params":{"query":"Python programming"}},"timestamp":"2025-10-22T10:30:02Z","seq":3,"stream_id":"1719000002000-0"}

id: 1719000005000-0
event: TOOL_OBSERVATION
data: {"workflow_id":"task_abc123","type":"TOOL_OBSERVATION","agent_id":"Ryogoku","message":"web_search completed","payload":{"success":true,"duration_ms":2800},"timestamp":"2025-10-22T10:30:05Z","seq":4,"stream_id":"1719000005000-0"}

id: 1719000006000-0
event: thread.message.completed
data: {"workflow_id":"task_abc123","agent_id":"simple-agent","response":"Python is a high-level programming language...","metadata":{"model_used":"claude-haiku-4-5-20251001","tokens_used":850,"cost_usd":0.003},"seq":5,"stream_id":"1719000006000-0"}

id: 1719000007000-0
event: WORKFLOW_COMPLETED
data: {"workflow_id":"task_abc123","type":"WORKFLOW_COMPLETED","message":"Workflow completed","timestamp":"2025-10-22T10:30:07Z","seq":6,"stream_id":"1719000007000-0"}
```

<Warning>
  **事件 ID 格式**：`id` 字段使用 Redis Stream ID（如 `1719000000000-0`），而**不是**简单整数。重连时必须使用这些精确的 ID 作为 `last_event_id` 参数。参见下方[重连](#reconnection)部分。
</Warning>

## WebSocket

### GET /api/v1/stream/ws

通过 WebSocket 进行双向流式传输。

#### 认证

网关通过仅使用请求头（`X-API-Key` 或 `Authorization`）来对 WebSocket 连接进行认证。浏览器在 WebSocket 握手期间无法设置自定义请求头。对于浏览器使用：

* 以 `GATEWAY_SKIP_AUTH=1` 本地运行，或
* 使用反向代理在转发到网关之前注入请求头。

服务器环境中基于请求头的示例：

Node (ws)：

```js theme={"dark"}
import WebSocket from 'ws';

const ws = new WebSocket('ws://localhost:8080/api/v1/stream/ws', {
  headers: { 'X-API-Key': 'sk_test_123456' },
});

ws.on('open', () => {
  ws.send(JSON.stringify({ type: 'subscribe', workflow_id: 'task_abc123' }));
});

ws.on('message', (msg) => {
  const data = JSON.parse(msg.toString());
  console.log('事件：', data.type, data.message);
});

ws.on('error', (err) => console.error('WebSocket 错误：', err));
```

Python (websockets)：

```python theme={"dark"}
import asyncio, json, websockets

async def main():
    async with websockets.connect(
        'ws://localhost:8080/api/v1/stream/ws',
        extra_headers={'X-API-Key': 'sk_test_123456'},
    ) as ws:
        await ws.send(json.dumps({'type': 'subscribe', 'workflow_id': 'task_abc123'}))
        async for message in ws:
            data = json.loads(message)
            print('事件：', data.get('type'), data.get('message'))

asyncio.run(main())
```

<Note>
  不支持通过查询字符串传递 API 密钥或通过网关的连接后"auth"消息。
</Note>

#### 消息类型

**客户端 → 服务器**：

```json theme={"dark"}
// 订阅工作流
{
  "type": "subscribe",
  "workflow_id": "task_abc123",
  "types": ["AGENT_THINKING", "TOOL_INVOKED"]
}

// 取消订阅
{
  "type": "unsubscribe",
  "workflow_id": "task_abc123"
}

// Ping（保活）
{
  "type": "ping"
}
```

**服务器 → 客户端**：

```json theme={"dark"}
// 事件
{
  "type": "AGENT_THINKING",
  "workflow_id": "task_abc123",
  "message": "分析查询...",
  "timestamp": "2025-10-22T10:30:00Z"
}

// Pong（保活响应）
{
  "type": "pong"
}
```

## OpenAI 兼容流式传输

Shannon 还在 `/v1/chat/completions` 提供 OpenAI 兼容的流式传输端点，将 Shannon 事件转换为标准 OpenAI `chat.completion.chunk` 格式。这允许您直接使用 OpenAI SDK 与 Shannon 交互。

有关 OpenAI 兼容 API 的完整文档，包括请求/响应模式、可用模型、Shannon 特有扩展（`shannon_events`）以及 SDK 使用示例，请参阅 [OpenAI 兼容 API 参考](/cn/api/rest/openai-compatible)。

## 事件类型

### 核心事件

| 事件类型                 | 描述      | 何时触发     |
| -------------------- | ------- | -------- |
| `WORKFLOW_STARTED`   | 工作流执行开始 | 任务开始     |
| `WORKFLOW_COMPLETED` | 工作流成功完成 | 任务结束（成功） |

### 智能体事件

| 事件类型              | 描述       | 载荷字段                 |
| ----------------- | -------- | -------------------- |
| `AGENT_THINKING`  | 智能体推理/规划 | `agent_id`、`message` |
| `AGENT_COMPLETED` | 智能体完成执行  | `agent_id`、`result`  |

### 工具事件

| 事件类型               | 描述        | 载荷字段            |
| ------------------ | --------- | --------------- |
| `TOOL_INVOKED`     | 工具执行已启动   | `tool`、`params` |
| `TOOL_OBSERVATION` | 智能体观察工具结果 | `tool`、`result` |

### LLM 事件

| 事件类型                       | 描述              | 载荷字段                                                                        |
| -------------------------- | --------------- | --------------------------------------------------------------------------- |
| `LLM_PROMPT`               | 发送给 LLM 的提示     | `message`、`payload.model`、`payload.provider`                                |
| `thread.message.delta`     | 流式文本片段          | `delta`、`agent_id`                                                          |
| `thread.message.completed` | 包含使用量的最终响应      | `response`、`metadata.model_used`、`metadata.tokens_used`、`metadata.cost_usd` |
| `LLM_PARTIAL`              | 原始 LLM 流式 token | `text`                                                                      |
| `LLM_OUTPUT`               | 原始 LLM 最终输出     | `text`                                                                      |

<Tip>
  **对于大多数集成场景**，建议监听 `thread.message.delta`（流式文本）和 `thread.message.completed`（包含使用量元数据的最终结果），而不是 `LLM_PARTIAL`/`LLM_OUTPUT`。
</Tip>

### 进度与系统事件

| 事件类型                | 描述         | 载荷字段                                                                                    |
| ------------------- | ---------- | --------------------------------------------------------------------------------------- |
| `PROGRESS`          | 进度更新       | `progress`、`message`                                                                    |
| `DATA_PROCESSING`   | 数据处理中      | `message`                                                                               |
| `WAITING`           | 等待资源       | `message`                                                                               |
| `ERROR_OCCURRED`    | 发生错误       | `error`、`severity`                                                                      |
| `ERROR_RECOVERY`    | 错误恢复尝试     | `message`                                                                               |
| `WORKSPACE_UPDATED` | 内存/上下文已更新  | `message`                                                                               |
| `BUDGET_THRESHOLD`  | 任务预算达到预警阈值 | `usage_percent`、`threshold_percent`、`tokens_used`、`tokens_budget`、`level`、`budget_type` |

### 流生命周期事件

| 事件类型         | 描述                    |
| ------------ | --------------------- |
| `STREAM_END` | 显式的流结束信号（该工作流不会再有新事件） |

<Note>
  在 SSE 中，`STREAM_END` 生命周期事件通过名为 `done` 的 SSE 事件发送，数据为纯文本 `[DONE]`（不是 JSON）。在 WebSocket 中，它会以普通 JSON 事件的形式出现，字段为 `"type": "STREAM_END"`。
</Note>

### 团队与审批

| 事件类型                   | 描述       |
| ---------------------- | -------- |
| `TEAM_RECRUITED`       | 组建多智能体团队 |
| `TEAM_RETIRED`         | 解散团队     |
| `TEAM_STATUS`          | 团队协调更新   |
| `ROLE_ASSIGNED`        | 分配智能体角色  |
| `DELEGATION`           | 任务已委托    |
| `DEPENDENCY_SATISFIED` | 依赖已解决    |
| `APPROVAL_REQUESTED`   | 需要人工批准   |
| `APPROVAL_DECISION`    | 审批决策已记录  |

## 代码示例

### Python with httpx (SSE)

```python theme={"dark"}
import httpx
import json

def stream_task_events(task_id: str, api_key: str):
    """使用 SSE 流式传输任务事件。"""
    url = f"http://localhost:8080/api/v1/stream/sse?workflow_id={task_id}"

    with httpx.stream(
        "GET",
        url,
        headers={"X-API-Key": api_key},
        timeout=None  # 无流式传输超时
    ) as response:
        for line in response.iter_lines():
            if line.startswith("data:"):
                data = json.loads(line[5:])  # 移除"data:"前缀
                yield data

# 用法
for event in stream_task_events("task_abc123", "sk_test_123456"):
    print(f"[{event.get('type')}] {event.get('message', '')}")

    if event.get('type') == 'WORKFLOW_COMPLETED':
        print("最终结果：", event.get('result'))
        break
```

### Python - 流式传输带事件筛选

```python theme={"dark"}
def stream_filtered_events(task_id: str, api_key: str, event_types: list):
    """仅流式传输特定事件类型。"""
    types_param = ",".join(event_types)
    url = f"http://localhost:8080/api/v1/stream/sse?workflow_id={task_id}&types={types_param}"

    with httpx.stream("GET", url, headers={"X-API-Key": api_key}) as response:
        for line in response.iter_lines():
            if line.startswith("data:"):
                yield json.loads(line[5:])

# 仅接收智能体思考和工具事件
for event in stream_filtered_events(
    "task_abc123",
    "sk_test_123456",
    ["AGENT_THINKING", "TOOL_INVOKED", "TOOL_OBSERVATION"]
):
    print(f"{event['type']}：{event.get('message', event.get('tool'))}")
```

### JavaScript/Node.js (SSE)

```javascript theme={"dark"}
const EventSource = require('eventsource');

function streamTaskEvents(taskId, apiKey) {
  const url = `http://localhost:8080/api/v1/stream/sse?workflow_id=${taskId}`;

  const eventSource = new EventSource(url, {
    headers: {
      'X-API-Key': apiKey
    }
  });

  eventSource.onmessage = (event) => {
    const data = JSON.parse(event.data);
    console.log(`[${data.type}] ${data.message || ''}`);

    if (data.type === 'WORKFLOW_COMPLETED') {
      console.log('最终结果：', data.result);
      eventSource.close();
    }
  };

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

  return eventSource;
}

// 用法
const stream = streamTaskEvents('task_abc123', 'sk_test_123456');

// 手动关闭
setTimeout(() => stream.close(), 60000);
```

### JavaScript/Node.js - WebSocket (ws)

```javascript theme={"dark"}
import WebSocket from 'ws';

function connectWebSocket(taskId, apiKey) {
  const ws = new WebSocket(
    `ws://localhost:8080/api/v1/stream/ws?workflow_id=${taskId}`,
    { headers: { 'X-API-Key': apiKey } }
  );

  ws.on('open', () => console.log('✓ 已连接'));

  ws.on('message', (msg) => {
    const data = JSON.parse(msg.toString());
    switch (data.type) {
      case 'AGENT_THINKING':
        console.log(`💭 ${data.message}`);
        break;
      case 'TOOL_INVOKED':
        console.log(`🔧 工具：${data.tool}`);
        break;
      case 'TOOL_OBSERVATION':
        console.log(`✓ 结果：${data.result}`);
        break;
      case 'WORKFLOW_COMPLETED':
        console.log(`✓ 完成：${data.result}`);
        ws.close();
        break;
      default:
        console.log(`[${data.type}] ${data.message || ''}`);
    }
  });

  ws.on('error', (err) => console.error('❌ 错误：', err));
  ws.on('close', () => console.log('连接已关闭'));

  return ws;
}

const ws = connectWebSocket('task_abc123', 'sk_test_123456');
```

### Go (SSE)

```go theme={"dark"}
package main

import (
    "bufio"
    "encoding/json"
    "fmt"
    "net/http"
    "strings"
)

type Event struct {
    Type      string                 `json:"type"`
    WorkflowID string                `json:"workflow_id"`
    Message   string                 `json:"message,omitempty"`
    Data      map[string]interface{} `json:",inline"`
}

func streamEvents(taskID, apiKey string) error {
    url := fmt.Sprintf(
        "http://localhost:8080/api/v1/stream/sse?workflow_id=%s",
        taskID,
    )

    req, _ := http.NewRequest("GET", url, nil)
    req.Header.Set("X-API-Key", apiKey)

    client := &http.Client{}
    resp, err := client.Do(req)
    if err != nil {
        return err
    }
    defer resp.Body.Close()

    scanner := bufio.NewScanner(resp.Body)

    for scanner.Scan() {
        line := scanner.Text()

        if strings.HasPrefix(line, "data:") {
            data := strings.TrimPrefix(line, "data:")

            var event Event
            json.Unmarshal([]byte(data), &event)

            fmt.Printf("[%s] %s\n", event.Type, event.Message)

            if event.Type == "WORKFLOW_COMPLETED" {
                fmt.Println("✓ 任务完成")
                break
            }
        }
    }

    return scanner.Err()
}

func main() {
    err := streamEvents("task_abc123", "sk_test_123456")
    if err != nil {
        fmt.Println("错误：", err)
    }
}
```

### Bash/curl (SSE)

```bash theme={"dark"}
#!/bin/bash

API_KEY="sk_test_123456"
TASK_ID="$1"

curl -N "http://localhost:8080/api/v1/stream/sse?workflow_id=$TASK_ID" \\
  -H "X-API-Key: $API_KEY" \\
  | while IFS= read -r line; do
    if [[ $line == data:* ]]; then
      # 从"data：{...}"提取 JSON
      JSON="${line#data:}"

      # 解析并显示
      TYPE=$(echo "$JSON" | jq -r '.type')
      MESSAGE=$(echo "$JSON" | jq -r '.message // ""')

      echo "[$(date +%T)] $TYPE：$MESSAGE"

      # 完成时退出
      if [[ "$TYPE" == "WORKFLOW_COMPLETED" ]]; then
        echo ""
        echo "$JSON" | jq -r '.result'
        break
      fi
    fi
  done
```

## 用例

### 1. 实时进度显示

```python theme={"dark"}
def display_progress(task_id: str, api_key: str):
    """向用户显示实时进度。"""
    print(f"任务 {task_id} 启动...")

    for event in stream_task_events(task_id, api_key):
        event_type = event.get('type')

        if event_type == 'AGENT_THINKING':
            print(f"💭 {event['message']}")
        elif event_type == 'TOOL_INVOKED':
            print(f"🔧 使用工具：{event['tool']}")
        elif event_type == 'TOOL_OBSERVATION':
            print(f"✓ 工具已完成")
        elif event_type == 'LLM_PARTIAL':
            print(event['text'], end='', flush=True)
        elif event_type == 'WORKFLOW_COMPLETED':
            print(f"\n\n✓ 完成！")
            return event['result']
```

### 2. 将所有事件记录到文件

```python theme={"dark"}
import json
from datetime import datetime

def log_events_to_file(task_id: str, api_key: str, log_file: str):
    """将所有事件记录到 JSON Lines 文件。"""
    with open(log_file, 'a') as f:
        for event in stream_task_events(task_id, api_key):
            # 添加时间戳
            event['logged_at'] = datetime.now().isoformat()

            # 以 JSON Lines 格式写入
            f.write(json.dumps(event) + '\n')
            f.flush()

            if event.get('type') == 'WORKFLOW_COMPLETED':
                break

log_events_to_file("task_abc123", "sk_test_123456", "task_events.jsonl")
```

### 3. 收集工具使用指标

```python theme={"dark"}
def collect_tool_metrics(task_id: str, api_key: str):
    """收集有关工具使用的指标。"""
    metrics = {
        "tools_invoked": 0,
        "tools_succeeded": 0,
        "tools_failed": 0,
        "tool_list": []
    }

    for event in stream_task_events(task_id, api_key):
        if event.get('type') == 'TOOL_INVOKED':
            metrics['tools_invoked'] += 1
            metrics['tool_list'].append(event['tool'])
        elif event.get('type') == 'TOOL_OBSERVATION':
            metrics['tools_succeeded'] += 1
        elif event.get('type') == 'ERROR_OCCURRED':
            metrics['tools_failed'] += 1
        elif event.get('type') == 'WORKFLOW_COMPLETED':
            break

    return metrics

metrics = collect_tool_metrics("task_abc123", "sk_test_123456")
print(f"使用的工具：{metrics['tool_list']}")
print(f"成功率：{metrics['tools_succeeded']}/{metrics['tools_invoked']}")
```

### 4. React UI 集成

```javascript theme={"dark"}
import { useState, useEffect } from 'react';

function TaskMonitor({ taskId, apiKey }) {
  const [events, setEvents] = useState([]);
  const [status, setStatus] = useState('connecting');

  useEffect(() => {
    // 浏览器 EventSource 不支持自定义请求头，使用 api_key 查询参数
    const eventSource = new EventSource(
      `http://localhost:8080/api/v1/stream/sse?workflow_id=${taskId}&api_key=${apiKey}`
    );

    eventSource.onopen = () => setStatus('connected');

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

      if (data.type === 'WORKFLOW_COMPLETED') {
        setStatus('completed');
        eventSource.close();
      }
    };

    eventSource.onerror = () => {
      setStatus('error');
      eventSource.close();
    };

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

  return (
    <div>
      <h3>状态：{status}</h3>
      <ul>
        {events.map((event, i) => (
          <li key={i}>
            <strong>{event.type}</strong>：{event.message || event.tool}
          </li>
        ))}
      </ul>
    </div>
  );
}
```

## 深度研究流式传输

深度研究任务通常需要 2-10 分钟。通过 Task API 使用 `context.force_research` 触发深度研究：

```bash theme={"dark"}
curl -X POST http://localhost:8080/api/v1/tasks/stream \
  -H "Content-Type: application/json" \
  -H "X-API-Key: sk_test_123456" \
  -d '{
    "query": "What are the latest developments in quantum computing?",
    "session_id": "session-123",
    "context": {"force_research": true}
  }'
```

通过 `context.research_strategy` 控制研究深度：

| 策略             | 深度 | 典型耗时     | 使用场景      |
| -------------- | -- | -------- | --------- |
| `quick`        | 低  | 30s-2min | 简单事实查询    |
| `standard`（默认） | 中  | 2-5min   | 通用研究      |
| `deep`         | 高  | 5-10min  | 复杂多面主题    |
| `academic`     | 最高 | 5-10min  | 引文密集的学术研究 |

<Warning>
  **Chat API vs Task API 深度研究**：Chat API（`/v1/chat/completions` 配合 `model: "shannon-deep-research"`）也可以触发深度研究，但其流式格式**不包含** SSE 事件 ID——因此无法重连。如果您的平台有连接时间限制（如 Vercel 5 分钟限制）或需要在页面刷新后恢复，**请使用 Task API**。
</Warning>

## 心跳

服务器每 10 秒发送一次 `: ping` SSE 注释以保持连接在代理和负载均衡器之间存活：

```
: ping
```

这是一条 SSE 注释（不是 JSON 消息）。如果您停止接收 ping，说明连接已断开——请立即重连。

## 重连

SSE 连接可能因网络问题、代理超时或平台限制（如 Vercel 免费版：5 分钟连接限制）而断开。Shannon 支持从断点处恢复。

**工作原理：**

1. 追踪每个接收到的 SSE 事件的 `id` 字段
2. 断连后，使用 `last_event_id` 参数重新连接
3. 服务器会重放该 ID 之后的所有事件（缓冲约 256 个事件，24 小时 TTL）

```bash theme={"dark"}
# 从断点处重连
curl -N "http://localhost:8080/api/v1/stream/sse?workflow_id=task_abc123&last_event_id=1719000002000-0" \
  -H "X-API-Key: sk_test_123456"
```

### 主动重连（推荐）

对于有连接时间限制的平台，建议在达到限制前主动断开：

```
1. 连接到 SSE
2. 启动一个 4 分钟的定时器（在 5 分钟限制前留出安全余量）
3. 定时器到期时：
   a. 记录最后接收到的事件 ID
   b. 关闭连接
   c. 立即使用 last_event_id=<last_id> 重新连接
4. 重复上述步骤，直到收到 WORKFLOW_COMPLETED 或 done 事件
```

### 页面刷新 / 回退方案

如果用户刷新了页面，且你仍保留 `workflow_id`：

1. 检查任务状态：`GET /api/v1/tasks/{workflow_id}`
2. 如果是 `TASK_STATUS_RUNNING` → 重新连接 SSE
3. 如果是 `TASK_STATUS_COMPLETED` → 直接显示响应中的结果
4. 如果是 `TASK_STATUS_FAILED` → 显示错误信息

### Python 重连示例

```python theme={"dark"}
import time

def stream_with_reconnect(task_id: str, api_key: str, max_retries: int = 5):
    """使用 last_event_id 实现自动重连的流式传输。"""
    last_id = None

    for attempt in range(max_retries):
        try:
            url = f"http://localhost:8080/api/v1/stream/sse?workflow_id={task_id}"
            if last_id:
                url += f"&last_event_id={last_id}"

            with httpx.stream("GET", url, headers={"X-API-Key": api_key}, timeout=None) as response:
                for line in response.iter_lines():
                    if line.startswith("id:"):
                        last_id = line[3:].strip()  # Track last event ID
                    elif line.startswith("data:"):
                        event = json.loads(line[5:])
                        yield event
                        if event.get('type') in ('WORKFLOW_COMPLETED', 'STREAM_END'):
                            return
        except Exception as e:
            if attempt < max_retries - 1:
                wait = min(2 ** attempt, 10)
                print(f"连接断开，{wait}s 后重连（last_id={last_id}）...")
                time.sleep(wait)
            else:
                raise
```

## 最佳实践

### 2. 实施超时

```python theme={"dark"}
import signal

def stream_with_timeout(task_id: str, api_key: str, timeout: int = 300):
    """具有超时的流式传输。"""
    def timeout_handler(signum, frame):
        raise TimeoutError("流式传输超时")

    signal.signal(signal.SIGALRM, timeout_handler)
    signal.alarm(timeout)

    try:
        for event in stream_task_events(task_id, api_key):
            yield event
            if event.get('type') == 'WORKFLOW_COMPLETED':
                break
    finally:
        signal.alarm(0)  # 取消闹钟
```

### 3. 客户端过滤事件

```python theme={"dark"}
def stream_specific_events(task_id: str, api_key: str, event_types: set):
    """客户端过滤事件。"""
    for event in stream_task_events(task_id, api_key):
        if event.get('type') in event_types:
            yield event

        if event.get('type') == 'WORKFLOW_COMPLETED':
            yield event
            break

# 仅接收智能体和工具事件
for event in stream_specific_events(
    "task_abc123",
    "sk_test_123456",
    {"AGENT_THINKING", "TOOL_INVOKED", "TOOL_OBSERVATION"}
):
    print(event)
```

### 4. 从最后一个事件恢复

```python theme={"dark"}
def stream_with_resume(task_id: str, api_key: str, last_event_id: str = None):
    """从特定事件恢复流式传输。"""
    url = f"http://localhost:8080/api/v1/stream/sse?workflow_id={task_id}"

    if last_event_id:
        url += f"&last_event_id={last_event_id}"

    # 流式传输事件...
```

说明：`last_event_id` 支持 Redis Stream ID（如 `1700000000000-0`）或数字序号（如 `42`）。当为数字时，重放规则为 `seq > last_event_id`。

## 比较：SSE vs WebSocket vs 轮询

| 功能             | SSE       | WebSocket | 轮询        |
| -------------- | --------- | --------- | --------- |
| **方向**         | 服务器 → 客户端 | 双向        | 客户端 → 服务器 |
| **协议**         | HTTP      | WebSocket | HTTP      |
| **自动重连**       | 是（浏览器）    | 否（手动）     | N/A       |
| **开销**         | 低         | 非常低       | 高         |
| **简洁性**        | 高         | 中         | 高         |
| **使用场景**       | 实时更新      | 交互式应用     | 简单状态      |
| **Shannon 支持** | ✅ 推荐      | ✅ 可用      | ⚠️ 不理想    |

### 何时使用每个

* **SSE**：大多数用例、实时监控、进度显示
* **WebSocket**：交互式应用、需要双向通信
* **轮询**（GET /api/v1/tasks/{id}）：传统系统、无流式传输支持

## 相关端点

<CardGroup cols={2}>
  <Card title="提交任务" icon="paper-plane" href="/cn/api/rest/submit-task">
    POST /api/v1/tasks
  </Card>

  <Card title="获取状态" icon="circle-info" href="/cn/api/rest/get-status">
    GET /api/v1/tasks/{id}
  </Card>

  <Card title="Python SDK" icon="python" href="/cn/sdk/python/quickstart">
    使用 client.stream()
  </Card>
</CardGroup>

## 注意

<Tip>
  **事件保留**：

  * **Redis**：所有事件存储 24 小时（实时流式传输）
  * **PostgreSQL**：关键事件存储 90 天（历史查询）
  * 如果连接断开，使用 `last_event_id` 恢复流式传输
</Tip>

<Warning>
  **连接限制**：

  * 每个 API 密钥最多 100 个并发流式传输连接
  * 5 分钟无活动超时（自动关闭连接）
  * 每个连接 1MB 缓冲区大小限制
  * 考虑在单个 WebSocket 上多路复用多个工作流
</Warning>