Wiring android kotlin flow streaming llm output into a mobile client turns a blocking request into a live token feed that users can read as it generates. This guide builds a complete pipeline using OkHttp, kotlinx.coroutines, and Jetpack Compose, targeting an OpenAI-compatible SSE endpoint. You will end up with a cancellable Flow that emits partial completions and survives configuration changes.
Step 1: Choose a streaming-compatible endpoint
OpenAI-style chat completions support stream: true, which returns Server-Sent Events (SSE). Each event is a data: line containing a JSON fragment with a delta field. A gateway such as n4n.ai exposes one OpenAI-compatible endpoint covering many models and performs automatic fallback when a provider is rate-limited, so the client code stays identical across model swaps.
A minimal request body looks like this:
{
"model": "gpt-4o-mini",
"messages": [{"role": "user", "content": "Explain coroutines"}],
"stream": true
}Verify the endpoint with curl before writing Kotlin:
curl -N -X POST https://api.example.com/v1/chat/completions \
-H "Authorization: Bearer $KEY" \
-H "Content-Type: application/json" \
-d '{"model":"gpt-4o-mini","messages":[{"role":"user","content":"hi"}],"stream":true}'You should see lines prefixed data: ending with data: [DONE].
Step 2: Add dependencies and permissions
Declare internet permission and use HTTPS. In AndroidManifest.xml:
<uses-permission android:name="android.permission.INTERNET" />In your module build.gradle.kts:
dependencies {
implementation("androidx.lifecycle:lifecycle-viewmodel-ktx:2.7.0")
implementation("org.jetbrains.kotlinx:kotlinx-coroutines-core:1.8.1")
implementation("org.jetbrains.kotlinx:kotlinx-coroutines-android:1.8.1")
implementation("com.squareup.okhttp3:okhttp:4.12.0")
implementation("org.jetbrains.kotlinx:kotlinx-serialization-json:1.6.3")
implementation("androidx.compose-ui:ui:1.6.4")
implementation("androidx.lifecycle:lifecycle-runtime-compose:2.7.0")
}Use kotlinx-serialization to parse chunks without reflection overhead.
Step 3: Define the streaming data models
The SSE payloads are partial. Model only what you consume:
import kotlinx.serialization.Serializable
@Serializable
data class ChatChunk(
val choices: List<Choice> = emptyList()
)
@Serializable
data class Choice(
val delta: Delta = Delta(),
val finish_reason: String? = null
)
@Serializable
data class Delta(
val content: String? = null
)The [DONE] sentinel arrives as a raw string, not JSON. Handle it outside serialization.
Step 4: Build the streaming HTTP client
OkHttp streams the body as a buffered source. Wrap it in flow and parse line by line. Use Dispatchers.IO for the network read.
import kotlinx.coroutines.flow.Flow
import kotlinx.coroutines.flow.flow
import kotlinx.coroutines.withContext
import okhttp3.MediaType.Companion.toMediaType
import okhttp3.Request
import okhttp3.RequestBody.Companion.toRequestBody
import okio.BufferedSource
import kotlinx.serialization.json.Json
fun streamChat(
client: OkHttpClient,
baseUrl: String,
apiKey: String,
prompt: String
): Flow<String> = flow {
val body = """
{"model":"gpt-4o-mini","messages":[{"role":"user","content":"$prompt"}],"stream":true}
""".trimIndent()
val request = Request.Builder()
.url("$baseUrl/v1/chat/completions")
.addHeader("Authorization", "Bearer $apiKey")
.addHeader("Content-Type", "application/json")
.post(body.toRequestBody("application/json".toMediaType()))
.build()
val response = client.newCall(request).execute()
if (!response.isSuccessful) {
throw IllegalStateException("HTTP ${response.code}")
}
val source: BufferedSource = response.body!!.source()
withContext(Dispatchers.IO) {
while (!source.exhausted()) {
val line = source.readUtf8Line() ?: continue
if (line.startsWith("data: ")) {
val data = line.removePrefix("data: ").trim()
if (data == "[DONE]") break
val chunk = Json.decodeFromString<ChatChunk>(data)
chunk.choices.firstOrNull()?.delta?.content?.let { emit(it) }
}
}
}
}This emits each content fragment as a String. Backpressure is handled by Flow’s default buffering; the collector suspends if it can’t keep up.
Step 5: Expose the stream from a ViewModel
Keep the coroutine scope tied to the ViewModel. Convert the raw token flow into UI state with stateIn or a simple MutableStateFlow.
class ChatViewModel(
private val client: OkHttpClient,
private val baseUrl: String,
private val apiKey: String
) : ViewModel() {
private val _messages = MutableStateFlow("")
val messages: StateFlow<String> = _messages.asStateFlow()
fun send(prompt: String) {
viewModelScope.launch {
_messages.value = ""
try {
streamChat(client, baseUrl, apiKey, prompt)
.catch { e -> _messages.value += "\n[error: ${e.message}]" }
.collect { token -> _messages.value += token }
} catch (e: CancellationException) {
throw e
} catch (e: Exception) {
_messages.value += "\n[error: ${e.message}]"
}
}
}
}Structured concurrency cancels the HTTP call automatically when the ViewModel clears or the user navigates away.
Step 6: Render tokens in Compose
Collect the StateFlow and display in a scrollable text box. The ViewModel drives the flow; Compose only observes.
@Composable
fun ChatScreen(viewModel: ChatViewModel) {
val text by viewModel.messages.collectAsState()
var input by remember { mutableStateOf("") }
Column(modifier = Modifier.padding(16.dp)) {
Text(
text = text,
modifier = Modifier
.weight(1f)
.verticalScroll(rememberScrollState())
)
Row {
TextField(
value = input,
onValueChange = { input = it },
modifier = Modifier.weight(1f)
)
Button(onClick = {
viewModel.send(input)
input = ""
}) { Text("Send") }
}
}
}The UI updates on every emitted token, giving the illusion of typing.
Step 7: Handle cancellation, timeouts, and retries
Mobile networks drop. Add a timeout and a single retry for transient errors:
import kotlinx.coroutines.flow.timeout
import kotlinx.coroutines.flow.retryWhen
import java.io.IOException
import java.net.SocketTimeoutException
import kotlin.time.Duration.Companion.seconds
fun streamChatWithPolicy(
client: OkHttpClient,
baseUrl: String,
apiKey: String,
prompt: String
): Flow<String> = streamChat(client, baseUrl, apiKey, prompt)
.timeout(30.seconds)
.retryWhen { cause, attempt ->
(cause is IOException || cause is SocketTimeoutException) && attempt < 1
}OkHttp cancels the underlying call when the coroutine is cancelled because the blocking read is inside withContext. Always use viewModelScope or rememberCoroutineScope so leaving the screen stops the stream. If your gateway honors provider cache hints, forward Cache-Control headers; n4n.ai forwards provider cache-control hints, which can cut latency on repeated system prompts.
Step 8: Verify success
You have three ways to confirm the android kotlin flow streaming llm integration works:
- Logcat: Log inside the
collectblock. You should see incremental non-empty strings, not one blob. - UI observation: Type a prompt, tap Send, and watch the text box fill without a full-screen spinner.
- Unit test: Use MockWebServer to serve canned SSE frames.
@Test
fun `emits tokens from sse`() = runTest {
val server = MockWebServer().apply {
enqueue(MockResponse().setBody(
"data: {\"choices\":[{\"delta\":{\"content\":\"Hello\"}}]}\ndata: [DONE]\n"
))
start()
}
val flow = streamChat(OkHttpClient(), server.url("/").toString(), "key", "hi")
val result = flow.toList()
assertEquals(listOf("Hello"), result)
server.shutdown()
}If the test passes and the device shows live text, the pipeline is production-ready. Tune buffer sizes with buffer(Channel.BUFFERED) only if you measure dropped frames under heavy token rates.
Pitfalls to avoid
Do not parse the whole body as JSON. SSE is line-delimited; readUtf8Line is mandatory. Never launch the stream in GlobalScope—process death leaks the connection. Do not call execute() on the main thread; the withContext(Dispatchers.IO) in Step 4 already moves the blocking read off the UI thread.
The android kotlin flow streaming llm pattern scales to multimodal inputs by extending the messages schema; the Flow contract stays the same. Build once, stream everywhere.