Claudio.Messages.Stream (Claudio v0.7.0)

View Source

Utilities for parsing and consuming Server-Sent Events (SSE) from streaming Messages API responses.

parse_events/1 emits [:claudio, :messages, :stream, :start | :stop] around each consumption (and the older [:claudio, :messages, :stream, :usage]); see the telemetry guide.

Event Types

The Messages API streaming responses include the following event types:

  • message_start - Initial message with empty content
  • content_block_start - Beginning of a content block
  • content_block_delta - Incremental content updates (text, JSON, thinking); read them with accumulate_text/1 / accumulate_thinking/1
  • content_block_stop - End of a content block
  • message_delta - Top-level message changes (usage updates)
  • message_stop - Stream completion
  • ping - Keep-alive events
  • error - Error events

Example

The simplest way to stream is to_response/2: print text as it arrives and still get usage, stop_reason and tool calls at the end.

{:ok, stream_response} =
  Claudio.Messages.create(client, Claudio.Messages.Request.enable_streaming(request))

{:ok, response} =
  Claudio.Messages.Stream.to_response(stream_response, on_text: &IO.write/1)

For lower-level control, enumerate the parsed events:

stream_response
|> Claudio.Messages.Stream.parse_events()
|> Stream.filter(&match?({:ok, %{event: "content_block_delta"}}, &1))
|> Enum.each(fn {:ok, event} ->
  IO.puts(event.data["delta"]["text"])
end)

Consume once

A streaming body can be read once, and only by the process that called Claudio.Messages.create/2: a second read waits forever, and a read from another process raises. Use the :on_text / :on_event options of to_response/2 instead of enumerating the response twice.

Summary

Types

event()
parsed_event()

Functions

accumulate_text(event_stream)

Emits the text of each text_delta event as a stream of text chunks.

accumulate_thinking(event_stream)

Emits {block_index, text} for every thinking_delta with non-empty text.

build_final_message(event_stream)

Accumulates all events and returns the final complete message.

filter_events(event_stream, event_types)

Filters stream to only specific event types.

parse_events(response)

Parses Server-Sent Events from a streaming response body.

to_response(source, opts \\ [])

Consumes a streaming response in one pass and returns the final Claudio.Messages.Response.

Types

event()

@type event() :: %{event: String.t(), data: map() | nil}

parsed_event()

@type parsed_event() :: {:ok, event()} | {:error, term()}

Functions

accumulate_text(event_stream)

@spec accumulate_text(Enumerable.t()) :: Enumerable.t()

Emits the text of each text_delta event as a stream of text chunks.

Enumerate them, or Enum.join/1 them for the full text.

Example

response
|> Stream.parse_events()
|> Stream.accumulate_text()
|> Enum.each(&IO.write/1)

accumulate_thinking(event_stream)

@spec accumulate_thinking(Enumerable.t()) :: Enumerable.t()

Emits {block_index, text} for every thinking_delta with non-empty text.

The index tells one thinking block from the next: with display: :updates each block is a separate progress note, and a block's first emission is the point the API docs say to treat it as an update. Empty deltas (display: :omitted), other deltas, other events and {:error, _} items emit nothing.

The index is nil only for a malformed frame with no "index" (the API always sends one); its text is still emitted rather than dropped.

Example

response
|> Stream.parse_events()
|> Stream.accumulate_thinking()
|> Enum.each(fn {index, text} -> IO.puts("[#{index}] #{text}") end)

build_final_message(event_stream)

@spec build_final_message(Enumerable.t()) :: {:ok, map()} | {:error, term()}

Accumulates all events and returns the final complete message.

Streamed tool input (input_json_delta chunks on tool_use, server_tool_use and mcp_tool_use blocks) is decoded into the block's "input". If a block's accumulated JSON is invalid — typically output cut off by max_tokens mid-call — the result is {:error, {:invalid_tool_input_json, index, partial_json}}.

Example

{:ok, final_message} =
  response
  |> Stream.parse_events()
  |> Stream.build_final_message()

filter_events(event_stream, event_types)

@spec filter_events(Enumerable.t(), [String.t()]) :: Enumerable.t()

Filters stream to only specific event types.

Example

response
|> Stream.parse_events()
|> Stream.filter_events(["content_block_delta", "message_stop"])

parse_events(response)

@spec parse_events(Req.Response.t() | Enumerable.t()) :: Enumerable.t()

Parses Server-Sent Events from a streaming response body.

Returns a Stream of {:ok, event} or {:error, reason} tuples.

Pass the whole %Req.Response{} from Claudio.Messages.create/2 to link the stream span to the create span; passing response.body still works, unlinked.

Emits [:claudio, :messages, :stream, :start | :stop] around each consumption of the returned stream; see the telemetry guide.

Example

response
|> Stream.parse_events()
|> Enum.to_list()

to_response(source, opts \\ [])

@spec to_response(
  Req.Response.t() | Enumerable.t(),
  keyword()
) :: {:ok, Claudio.Messages.Response.t()} | {:error, term()}

Consumes a streaming response in one pass and returns the final Claudio.Messages.Response.

The simplest way to stream: print text as it arrives and still get usage, stop_reason and tool calls at the end.

{:ok, response} =
  Claudio.Messages.Stream.to_response(stream_response, on_text: &IO.write/1)

response.usage.output_tokens

Options

  • :on_text — called with each text delta, in order.

  • :on_event — called with each parsed event ({:ok, event} | {:error, reason}).

A streaming body can be read once, and only by the process that called Claudio.Messages.create/2: a second read waits forever, and a read from another process raises. Use :on_text / :on_event instead of enumerating the response twice.

An SSE error event is returned as {:error, %Claudio.APIError{}}; other failures are as for build_final_message/1. That error's status_code is 200 (the stream's HTTP status), so match on its type.