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

# Event Streaming

> Real-time event streaming with the Python SDK

## Overview

Use SSE for simple, reliable real-time updates. Optional WebSocket is available for environments that require WS.

## SSE (recommended)

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

h = client.submit_task(
  "Draft a ~120-word executive summary of Q4 revenue drivers and risks. Stream partial output; end the last line with 'Conclusion:'."
)
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
```

Resume from last event:

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

Notes:

* `last_event_id` accepts either a Redis stream ID (preferred) or a numeric sequence. The SDK exposes `event.id` which chooses `stream_id` when present or falls back to `seq`.

## WebSocket (optional)

Requires `pip install websockets`.

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

h = client.submit_task(
  "Produce a ~120-word brief on recent renewable energy policy changes and their impact on costs. Stream partial output."
)
for e in client.stream_ws(h.workflow_id):
  print(e.type, e.message)
  if e.type == "WORKFLOW_COMPLETED":
    break
```

## Common Event Types

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

Tip: Filter using `EventType` enums or raw strings.