> ## 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.

# 事件流式传输

> 使用 Python SDK 进行实时事件流式传输

## 概述

推荐使用 SSE 获取简单可靠的实时更新。若环境需要 WS，可使用可选的 WebSocket 辅助方法。

## SSE（推荐）

```python theme={null}
from shannon import ShannonClient, EventType
client = ShannonClient()

h = client.submit_task(
  "起草一份约 120 字的关于 Q4 营收驱动与风险的执行摘要。流式传输部分输出；最后一行以‘结论：’结尾。"
)
for e in client.stream(h.workflow_id, types=[EventType.LLM_OUTPUT, EventType.WORKFLOW_COMPLETED]):
  print(f"{e.type}: {e.message}")
  if e.type == EventType.WORKFLOW_COMPLETED:
    break
```

从上一个事件恢复：

```python theme={null}
last_id = None
for e in client.stream(h.workflow_id, last_event_id=last_id):
  # ...
  last_id = e.id
```

注意：

* `last_event_id` 接受 Redis 流 ID（优先）或数字序号。SDK 的 `event.id` 会在存在时选择 `stream_id`，否则回退到 `seq`。

## WebSocket（可选）

需要 `pip install websockets`。

```python theme={null}
from shannon import ShannonClient
client = ShannonClient()

h = client.submit_task(
  "生成一份约 120 字的关于近期可再生能源政策变化及其对成本影响的简报。流式传输部分输出。"
)
for e in client.stream_ws(h.workflow_id):
  print(e.type, e.message)
  if e.type == "WORKFLOW_COMPLETED":
    break
```

## 常见事件类型

* WORKFLOW\_STARTED / WORKFLOW\_COMPLETED
* LLM\_PROMPT / LLM\_PARTIAL / LLM\_OUTPUT
* TOOL\_INVOKED / TOOL\_OBSERVATION
* APPROVAL\_REQUESTED / APPROVAL\_DECISION
* ERROR\_OCCURRED

提示：使用 `EventType` 枚举或原始字符串进行过滤。