import gleam/dynamic/decode.{type Decoder} import gleam/json import gleam/list import gleam/option.{type Option} import gleam/string import claude/json/decode as api_decode import claude/types/content.{type ContentBlock, Text, ToolUse} import claude/types/message.{type Message, type StopReason} /// A delta in a streaming content block. pub type Delta { TextDelta(String) InputJsonDelta(String) ThinkingDelta(String) } /// Usage information in a message_delta event. pub type DeltaUsage { DeltaUsage(output_tokens: Int) } /// A streaming event from the Anthropic Messages API (SSE). pub type StreamEvent { MessageStart(message: Message) ContentBlockStart(index: Int, content_block: ContentBlock) ContentBlockDelta(index: Int, delta: Delta) ContentBlockStop(index: Int) MessageDelta(stop_reason: Option(StopReason), usage: DeltaUsage) MessageStop Ping UnknownEvent(event_type: String, data: String) } /// Parse a full SSE text payload into a list of StreamEvents. /// /// SSE events are separated by blank lines ("\n\n"). Each event block /// contains an `event:` line and a `data:` line. pub fn parse_sse(text: String) -> List(StreamEvent) { text |> string.split("\n\n") |> list.filter(fn(block) { !string.is_empty(string.trim(block)) }) |> list.filter_map(parse_block) } /// Parse a single SSE block into a StreamEvent. /// Returns Error(Nil) if the block cannot be parsed. fn parse_block(block: String) -> Result(StreamEvent, Nil) { let lines = string.split(block, "\n") let event_type = find_field(lines, "event:") let data = find_field(lines, "data:") case event_type, data { Ok(evt), Ok(d) -> Ok(parse_event(evt, d)) Ok(evt), Error(_) -> case evt { "ping" -> Ok(Ping) _ -> Ok(UnknownEvent(event_type: evt, data: "")) } _, _ -> Error(Nil) } } /// Extract the value after a field prefix (e.g. "event:" or "data:") from SSE lines. fn find_field(lines: List(String), prefix: String) -> Result(String, Nil) { list.find_map(lines, fn(line) { let trimmed = string.trim(line) case string.starts_with(trimmed, prefix) { True -> { let value = string.drop_start(trimmed, string.length(prefix)) Ok(string.trim(value)) } False -> Error(Nil) } }) } /// Parse a single event given its type and JSON data string. pub fn parse_event(event_type: String, data: String) -> StreamEvent { case event_type { "message_start" -> parse_message_start(data) "content_block_start" -> parse_content_block_start(data) "content_block_delta" -> parse_content_block_delta(data) "content_block_stop" -> parse_content_block_stop(data) "message_delta" -> parse_message_delta(data) "message_stop" -> MessageStop "ping" -> Ping _ -> UnknownEvent(event_type: event_type, data: data) } } // --- Individual event parsers --- fn parse_message_start(data: String) -> StreamEvent { let decoder = { use message <- decode.field("message", api_decode.message_decoder()) decode.success(MessageStart(message: message)) } case json.parse(data, decoder) { Ok(event) -> event Error(_) -> UnknownEvent(event_type: "message_start", data: data) } } fn parse_content_block_start(data: String) -> StreamEvent { let decoder = { use index <- decode.field("index", decode.int) use block <- decode.field("content_block", content_block_start_decoder()) decode.success(ContentBlockStart(index: index, content_block: block)) } case json.parse(data, decoder) { Ok(event) -> event Error(_) -> UnknownEvent(event_type: "content_block_start", data: data) } } fn parse_content_block_delta(data: String) -> StreamEvent { let decoder = { use index <- decode.field("index", decode.int) use delta <- decode.field("delta", delta_decoder()) decode.success(ContentBlockDelta(index: index, delta: delta)) } case json.parse(data, decoder) { Ok(event) -> event Error(_) -> UnknownEvent(event_type: "content_block_delta", data: data) } } fn parse_content_block_stop(data: String) -> StreamEvent { let decoder = { use index <- decode.field("index", decode.int) decode.success(ContentBlockStop(index: index)) } case json.parse(data, decoder) { Ok(event) -> event Error(_) -> UnknownEvent(event_type: "content_block_stop", data: data) } } fn parse_message_delta(data: String) -> StreamEvent { let decoder = { use stop_reason <- decode.subfield( ["delta", "stop_reason"], decode.optional(api_decode.stop_reason_decoder()), ) use output_tokens <- decode.subfield( ["usage", "output_tokens"], decode.int, ) decode.success(MessageDelta( stop_reason: stop_reason, usage: DeltaUsage(output_tokens: output_tokens), )) } case json.parse(data, decoder) { Ok(event) -> event Error(_) -> UnknownEvent(event_type: "message_delta", data: data) } } // --- Decoders --- /// Decoder for a content block in content_block_start events. /// Handles "text" and "tool_use" types. fn content_block_start_decoder() -> Decoder(ContentBlock) { use block_type <- decode.field("type", decode.string) case block_type { "text" -> { use text <- decode.field("text", decode.string) decode.success(Text(text: text, citations: [])) } "tool_use" -> { use id <- decode.field("id", decode.string) use name <- decode.field("name", decode.string) decode.success(ToolUse(id: id, name: name, input: "")) } _ -> decode.success(Text(text: "", citations: [])) } } /// Decoder for the delta object in content_block_delta events. fn delta_decoder() -> Decoder(Delta) { use delta_type <- decode.field("type", decode.string) case delta_type { "text_delta" -> { use text <- decode.field("text", decode.string) decode.success(TextDelta(text)) } "input_json_delta" -> { use partial_json <- decode.field("partial_json", decode.string) decode.success(InputJsonDelta(partial_json)) } "thinking_delta" -> { use thinking <- decode.field("thinking", decode.string) decode.success(ThinkingDelta(thinking)) } _ -> decode.success(TextDelta("")) } }