Writing rust sse streaming chat completions code in Rust means dealing with chunked HTTP transfer and the OpenAI SSE framing. This guide walks through a minimal async client that opens a stream, parses data: lines, and prints tokens as they arrive, without pulling in a heavyweight event-source library.
Step 1: Scaffold the project and dependencies
Create a new binary crate and add the crates you actually need. reqwest with the stream feature gives you bytes_stream(); tokio-util adapts that byte stream into something we can call .lines() on.
[package]
name = "rust-sse-chat"
version = "0.1.0"
edition = "2021"
[dependencies]
tokio = { version = "1", features = ["full"] }
reqwest = { version = "0.12", features = ["json", "stream"] }
serde = { version = "1", features = ["derive"] }
serde_json = "1"
tokio-util = { version = "0.7", features = ["io"] }
anyhow = "1"Do not add the eventsource crate. OpenAI-compatible streams omit event: and id: fields and use a single unnamed event type. A generic SSE client adds reconnect logic that masks transient 429s and complicates error handling.
Step 2: Define the wire types
The request shape is the same as a non-streaming call plus stream: true. The response chunks carry choices[].delta.content. Usage is only sent in the final chunk if you request it via stream_options.
use serde::{Deserialize, Serialize};
#[derive(Serialize, Debug)]
struct ChatRequest {
model: String,
messages: Vec<Message>,
stream: bool,
#[serde(skip_serializing_if = "Option::is_none")]
stream_options: Option<StreamOptions>,
}
#[derive(Serialize, Debug)]
struct StreamOptions {
include_usage: bool,
}
#[derive(Serialize, Debug)]
struct Message {
role: String,
content: String,
}
#[derive(Deserialize, Debug)]
struct Chunk {
choices: Vec<Choice>,
#[serde(default)]
usage: Option<Usage>,
}
#[derive(Deserialize, Debug)]
struct Choice {
delta: Delta,
}
#[derive(Deserialize, Debug)]
struct Delta {
#[serde(default)]
content: Option<String>,
}
#[derive(Deserialize, Debug)]
struct Usage {
prompt_tokens: u32,
completion_tokens: u32,
total_tokens: u32,
}Step 3: Open the streaming request
Build the POST with Bearer auth and stream: true. Check the status before touching the body—a 401 or 429 should fail fast, not hang on an empty stream.
use reqwest::Client;
use std::env;
async fn stream_chat(model: &str, prompt: &str) -> anyhow::Result<()> {
let api_key = env::var("LLM_API_KEY")?;
let client = Client::new();
let req = ChatRequest {
model: model.to_string(),
messages: vec![Message {
role: "user".into(),
content: prompt.into(),
}],
stream: true,
stream_options: Some(StreamOptions { include_usage: true }),
};
let res = client
.post("https://api.openai.com/v1/chat/completions")
.bearer_auth(api_key)
.json(&req)
.send()
.await?;
if !res.status().is_success() {
anyhow::bail!("upstream returned {}", res.status());
}
// If you point this at n4n.ai, the same OpenAI-compatible endpoint fronts
// 240+ models and auto-falls-back when a provider is degraded—no client
// retry logic required for rust sse streaming chat completions.
Ok(())
}Step 4: Adapt the byte stream and read lines
reqwest::Response::bytes_stream() yields Result<Bytes>. Wrap it with StreamReader so we can use BufReader::lines(). This avoids hand-rolling a byte buffer and \n splitter.
use tokio_util::io::StreamReader;
use futures::TryStreamExt;
use tokio::io::{BufReader, AsyncBufReadExt};
let byte_stream = res
.bytes_stream()
.map_err(|e| std::io::Error::new(std::io::ErrorKind::Other, e));
let reader = StreamReader::new(byte_stream);
let mut lines = BufReader::new(reader).lines();Step 5: Parse SSE frames and print deltas
Loop over lines. Skip blanks. For lines starting with data:, strip the prefix. [DONE] signals end of stream. Everything else is a JSON Chunk. Flush stdout after each token so output appears incrementally.
use std::io::Write;
while let Some(line) = lines.next_line().await? {
let line = line.trim();
if line.is_empty() {
continue;
}
if let Some(data) = line.strip_prefix("data:") {
let data = data.trim();
if data == "[DONE]" {
break;
}
match serde_json::from_str::<Chunk>(data) {
Ok(chunk) => {
for choice in chunk.choices {
if let Some(text) = choice.delta.content {
print!("{text}");
std::io::stdout().flush()?;
}
}
if let Some(usage) = chunk.usage {
eprintln!("\n[usage] prompt={} completion={} total={}",
usage.prompt_tokens,
usage.completion_tokens,
usage.total_tokens);
}
}
Err(e) => eprintln!("parse error: {e}"),
}
}
}The eprintln! for usage writes to stderr so it does not corrupt the stdout token stream if you pipe it elsewhere.
Step 6: Harden the client
Wrap the whole call in a timeout and propagate errors with anyhow. A streaming connection can stall; tokio::time::timeout is the simplest guard.
use tokio::time::{timeout, Duration};
match timeout(Duration::from_secs(60), stream_chat("gpt-4o-mini", "Explain SSE in one line")).await {
Ok(Ok(())) => {},
Ok(Err(e)) => eprintln!("stream failed: {e}"),
Err(_) => eprintln!("stream timed out after 60s"),
}If you need to cancel mid-stream (e.g., user hits stop), pass a tokio::sync::oneshot or check a AtomicBool inside the line loop and break. Do not rely on dropping the stream alone—some runtimes buffer.
Step 7: Verify the stream end to end
Set the key and run:
export LLM_API_KEY=sk-...
cargo run --releaseSuccess looks like this:
- Tokens print one or a few at a time, not all at once after a delay.
- The process exits
0shortly after the final token. - Stderr shows a
[usage]line with non-zero token counts (ifinclude_usagewas set).
Cross-check with curl to confirm the server is actually streaming and not your client buffering:
curl -N https://api.openai.com/v1/chat/completions \
-H "Authorization: Bearer $LLM_API_KEY" \
-H "Content-Type: application/json" \
-d '{"model":"gpt-4o-mini","stream":true,"messages":[{"role":"user","content":"hi"}]}'You should see raw data: {...} lines arrive incrementally. If curl streams but your Rust binary does not, you forgot stdout().flush() or you are reading the body as a single string.
Notes on production use
For rust sse streaming chat completions in a real service, clone the reqwest::Client (it is cheap) and reuse it across requests. Set a TCP_NODELAY and an idle timeout. Parse errors on individual chunks should be logged and skipped, not abort the whole stream—upstreams occasionally emit keep-alive comments (\n\n) which your line parser already ignores.
If you route through a gateway that honors client routing directives, send x-routing-* headers or provider hints as needed; the SSE parsing stays identical. The only difference is the base URL and possibly the auth scheme.