anthropic/streaming/sse

Server-Sent Events (SSE) parser for Anthropic streaming responses

This module provides parsing utilities for SSE format used by the Anthropic streaming API. It handles multi-line data fields, event types, keepalives, and malformed events gracefully.

Types

SseError

</>

SSE parsing error

pub type SseError {
  InvalidField(line: String)
  EmptyEvent
  InvalidRetry(value: String)
}

Constructors

```
InvalidField(line: String)
```
Invalid field format
```
EmptyEvent
```
Empty event (no data)
```
InvalidRetry(value: String)
```
Invalid retry value

SseEvent

</>

A parsed SSE event

pub type SseEvent {
  SseEvent(
    event_type: option.Option(String),
    data: option.Option(String),
    id: option.Option(String),
    retry: option.Option(Int),
  )
}

Constructors

```
SseEvent(
  event_type: option.Option(String),
  data: option.Option(String),
  id: option.Option(String),
  retry: option.Option(Int),
)
```
Arguments

event_type

The event type (e.g., “message_start”, “content_block_delta”)

data

The event data (JSON string)

id

Event ID (rarely used)

retry

Retry interval in milliseconds (rarely used)

SseParseResult

</>

Result of parsing SSE data

pub type SseParseResult {
  SseParseResult(events: List(SseEvent), state: SseParserState)
}

Constructors

SseParseResult(events: List(SseEvent), state: SseParserState)

Arguments

events: Parsed events
state: Updated parser state

SseParserState

</>

Parser state for accumulating SSE events

pub type SseParserState {
  SseParserState(
    current_event_type: option.Option(String),
    current_data: List(String),
    current_id: option.Option(String),
    current_retry: option.Option(Int),
    buffer: String,
  )
}

Constructors

SseParserState(
  current_event_type: option.Option(String),
  current_data: List(String),
  current_id: option.Option(String),
  current_retry: option.Option(Int),
  buffer: String,
)

Arguments

current_event_type: Current event type being parsed
current_data: Accumulated data lines
current_id: Current event ID
current_retry: Current retry value
buffer: Buffer for incomplete lines

Values

flush

</>

pub fn flush(state: SseParserState) -> Result(SseEvent, SseError)

Flush any remaining data in the buffer as a final event

get_data

</>

pub fn get_data(event: SseEvent) -> String

Get the data as a string, defaulting to empty string

get_event_type

</>

pub fn get_event_type(event: SseEvent) -> String

Get the event type as a string, defaulting to “message” if not specified

is_keepalive

</>

pub fn is_keepalive(event: SseEvent) -> Bool

Check if an SSE event is a keepalive/ping

new_parser_state

</>

pub fn new_parser_state() -> SseParserState

Create a new parser state

parse_chunk

</>

pub fn parse_chunk(
  state: SseParserState,
  chunk: String,
) -> SseParseResult

Parse a chunk of SSE data, returning parsed events and updated state

This function handles partial events that span multiple chunks by maintaining a buffer in the parser state.

parse_event

</>

pub fn parse_event(
  event_block: String,
) -> Result(SseEvent, SseError)

Parse a single SSE event from a complete event block (lines separated by \n)

parse_event_lines

</>

pub fn parse_event_lines(
  lines: List(String),
) -> Result(SseEvent, SseError)

Parse SSE event from a list of lines

parse_line

</>

pub fn parse_line(
  state: SseParserState,
  line: String,
) -> SseParserState

Parse a single line and update parser state

reset_event_state

</>

pub fn reset_event_state(state: SseParserState) -> SseParserState

Reset parser state for next event (keeps buffer)

sse_event

</>

pub fn sse_event(
  event_type: option.Option(String),
  data: option.Option(String),
) -> SseEvent

Create an SSE event (primarily for testing)

sse_event_full

</>

pub fn sse_event_full(
  event_type: option.Option(String),
  data: option.Option(String),
  id: option.Option(String),
  retry: option.Option(Int),
) -> SseEvent

Create an SSE event with all fields