MCP.Transport.SSE (MCP Elixir SDK v1.1.0)

Copy Markdown View Source

Server-Sent Events (SSE) encoding and decoding utilities.

Used by the Streamable HTTP transport for streaming JSON-RPC messages over SSE connections.

SSE Format

SSE events consist of field lines separated by newlines, with events separated by blank lines:

event: message
id: 1
data: {"jsonrpc":"2.0",...}

Fields:

  • event: — event type (default: "message")
  • id: — event ID for resumability
  • data: — event payload (JSON-RPC message)
  • retry: — reconnection interval in milliseconds

Summary

Types

event()

Functions

decode_event(text)

Decodes an SSE event string into an event map.

encode_event(event)

Encodes an SSE event map into a string suitable for streaming.

encode_message(message, opts \\ [])

Encodes a JSON-RPC message as an SSE event.

feed(buffer, data)

Feeds data to a stream parser and returns any complete events.

new_parser()

Creates a new stream parser for incrementally parsing SSE events from chunked data.

Types

event()

@type event() :: %{
  optional(:event) => String.t(),
  optional(:id) => String.t(),
  optional(:data) => String.t(),
  optional(:retry) => non_neg_integer()
}

Functions

decode_event(text)

@spec decode_event(String.t()) :: {:ok, event()} | {:error, term()}

Decodes an SSE event string into an event map.

Parses the standard SSE fields (event, id, data, retry) from the event text. Multiple data: lines are joined with newlines.

Returns {:ok, event} or {:error, reason}.

encode_event(event)

@spec encode_event(event()) :: String.t()

Encodes an SSE event map into a string suitable for streaming.

Examples

iex> MCP.Transport.SSE.encode_event(%{event: "message", data: ~s({"jsonrpc":"2.0"})})
"event: message\ndata: {\"jsonrpc\":\"2.0\"}\n\n"

iex> MCP.Transport.SSE.encode_event(%{id: "42", data: "hello"})
"id: 42\ndata: hello\n\n"

encode_message(message, opts \\ [])

@spec encode_message(
  map(),
  keyword()
) :: String.t()

Encodes a JSON-RPC message as an SSE event.

Convenience function that JSON-encodes the message and wraps it in an SSE event with the "message" event type.

Options

  • :id — event ID for resumability
  • :event — event type (default: "message")

feed(buffer, data)

@spec feed(binary(), binary()) :: {[event()], binary()}

Feeds data to a stream parser and returns any complete events.

Returns {events, new_parser_state} where events is a list of decoded event maps.

new_parser()

@spec new_parser() :: binary()

Creates a new stream parser for incrementally parsing SSE events from chunked data.

Returns an initial parser state. Feed data to it using feed/2.

Example

parser = MCP.Transport.SSE.new_parser()
{events, parser} = MCP.Transport.SSE.feed(parser, chunk1)
{events, parser} = MCP.Transport.SSE.feed(parser, chunk2)