Server-Sent Events in Ktor Client

Server-Sent Events (SSE) is a technology that allows a server to continuously push events to a client over an HTTP connection. It's particularly useful in cases where the server needs to send event-based updates without requiring the client to repeatedly poll the server.

The SSE plugins supported by Ktor provide a straightforward method for creating a one-way connection between the server and the client.

Add dependencies

SSE only requires the ktor-client-core artifact and doesn't need any specific dependencies.

Install SSE

To install the SSE plugin, pass it to the install function inside a client configuration block:

            import io.ktor.client.*
            import io.ktor.client.engine.cio.*
            import io.ktor.client.plugins.sse.*

            //...
            val client = HttpClient(CIO) {
                install(SSE)
            }
        

Configure the SSE plugin

You can optionally configure the SSE plugin within the install block by setting the supported properties of the SSEConfig class.

SSE reconnect

To enable automatic reconnection, set maxReconnectionAttempts to a value greater than 0. You can also configure the delay between attempts using reconnectionTime:

                install(SSE) {
                    maxReconnectionAttempts = 4
                    reconnectionTime = 2.seconds
                }
            

If the connection to the server is lost, the client will wait for the specified reconnectionTime before attempting to reconnect. It will make up to the specified maxReconnectionAttempts to reestablish the connection.

Filter events

In the following example, the SSE plugin is installed into the HTTP client and configured to include events that only contain comments and those that contain only the retry field in the incoming flow:

        install(SSE) {
            showCommentEvents()
            showRetryEvents()
        }

Response buffering

SSE responses are streaming by nature, which makes capturing the full body impractical. You can enable a diagnostic buffer to safely retrieve the response body when an SSE stream fails. The buffer only contains data that has already been processed (no re-reading from the network) and is intended for logging and error analysis in case of failures.

                install(SSE) {
                    bufferPolicy = SSEBufferPolicy.LastEvents(10)
                }
            

You can also configure the buffer per call:

                client.sse(url, {
                    bufferPolicy(SSEBufferPolicy.All)
                }) {
                    // ...
                }
            

Buffer policies

The SSEBufferPolicy type provides several strategies for storing processed SSE data. These policies control how much of the stream is retained in memory and made available in case of errors.

Off (default): No buffering.
LastLines(n): Keeps the last n lines.
LastEvent: Keeps the last completed SSE event.
LastEvents(n): Keeps the last n completed SSE events.
All: Keeps all processed events so far.
Use with caution for long-lived streams.

On failure, you can access the buffer by using response?.bodyAsText() without re-reading from the network.

Handle SSE sessions

A client's SSE session is represented by the ClientSSESession interface. This interface exposes the API that allows you to receive server-sent events from a server.

Access an SSE session

The HttpClient allows you to get access to an SSE session in one of the following ways:

The sse() function creates the SSE session and allows you to act on it.
The sseSession() function allows you to open an SSE session.

To specify the URL endpoint, you can choose from two options:

Use the urlString parameter to specify the whole URL as a string.
Use the schema, host, port, and path parameters to specify the protocol scheme, domain name, port number and path name respectively.

                runBlocking {
                    client.sse(host = "127.0.0.1", port = 8080, path = "/events") {
                        // this: ClientSSESession
                    }
                }
            

Optionally, the following parameters are available to configure the connection:

reconnectionTime: Sets the reconnection delay.
showCommentEvents: Specifies whether to show events that contain only comments in the incoming flow.
showRetryEvents: Specifies whether to show events that contain only the retry field in the incoming flow.
deserialize: A deserializer function to transform the data field of the TypedServerSentEvent into an object. For more information, see Deserialization.

SSE session block

Within the lambda argument, you have access to the ClientSSESession context. The following properties are available within the block:

call: The associated HttpClientCall that originated the session.
incoming: An incoming server-sent events flow.

The example below creates a new SSE session with the events endpoint, reads events through the incoming property and prints the received ServerSentEvent.

fun main() {
    val client = HttpClient {
        install(SSE) {
            showCommentEvents()
            showRetryEvents()
        }
    }
    runBlocking {
        client.sse(host = "0.0.0.0", port = 8080, path = "/events") {
            while (true) {
                incoming.collect { event ->
                    println("Event from server:")
                    println(event)
                }
            }
        }
    }
}

For the full example, see client-sse.

Deserialization

The SSE plugin supports deserialization of server-sent events into type-safe Kotlin objects. This feature is particularly useful when working with structured data from the server.

To enable deserialization, provide a custom deserialization function using the deserialize parameter on an SSE access function and use the ClientSSESessionWithDeserialization class to handle the deserialized events.

Here's an example using kotlinx.serialization to deserialize JSON data:

        client.sse({
            url("http://localhost:8080/serverSentEvents")
        }, deserialize = {
                typeInfo, jsonString ->
            val serializer = Json.serializersModule.serializer(typeInfo.kotlinType!!)
            Json.decodeFromString(serializer, jsonString)!!
        }) { // `this` is `ClientSSESessionWithDeserialization`
            incoming.collect { event: TypedServerSentEvent<String> ->
                when (event.event) {
                    "customer" -> {
                        val customer: Customer? = deserialize<Customer>(event.data)
                    }
                    "product" -> {
                        val product: Product? = deserialize<Product>(event.data)
                    }
                }
            }
        }

For the full example, see client-sse.

20 October 2025