Streaming OpenAI Responses in Phoenix LiveView

If you’ve ever used ChatGPT, you know the feeling: the answer types itself out, word by word, instead of making you stare at a spinner for ten seconds. That token-by-token effect isn’t a gimmick — it’s the single biggest perceived-performance win you can add to an AI feature. And it turns out streaming OpenAI responses in Phoenix LiveView is one of the most natural things you’ll ever build in Elixir.

Most tutorials on this topic are written for Python or Node, where you have to bolt on WebSockets, a client-side state machine, and a pile of useEffect hooks just to push tokens to the browser. In Phoenix, the stateful WebSocket connection already exists — it’s the LiveView itself. You stream tokens from OpenAI on the server, send/2 them into your LiveView process, and let the diffing engine paint them onto the page. No JSON API, no frontend framework, no client state.

This guide walks through the whole thing end to end: calling the OpenAI streaming API with Req, parsing the server-sent events, pushing tokens into a LiveView without blocking it, rendering them smoothly, and handling the errors and cancellations that real users will throw at you.

Why LiveView is the right tool for streaming AI

A normal HTTP request is a dead end for streaming: the client asks, the server answers once, the connection closes. To stream, the traditional stacks open a second long-lived channel (SSE or a WebSocket) and reconcile it with the DOM by hand.

LiveView inverts this. Your page is already backed by a persistent process on the server, connected over a WebSocket that Phoenix manages for you. That means:

• State lives on the server. The growing assistant message is just a string in your socket assigns. You append to it and re-render.
• No API layer. You’re not serializing tokens to JSON and fetching them from JavaScript. You mutate assigns; LiveView sends a minimal diff.
• Backpressure and lifecycle are free. If the user navigates away, the LiveView process dies and your streaming task can be torn down with it.

The one rule to internalize: never block the LiveView process. A LiveView handles one message at a time. If you sit in a loop reading an HTTP stream inside handle_event, the whole page freezes — no clicks, no navigation, nothing. So we run the OpenAI call in a separate task and send tokens back as messages.

Setting up: Req and your API key

We’ll use Req, the modern Elixir HTTP client, because it exposes a clean callback for streaming response bodies. Add it to mix.exs:

defp deps do
  [
    {:req, "~> 0.5"}
  ]
end

Run mix deps.get, then keep your API key out of source with runtime config in config/runtime.exs:

config :my_app, :openai,
  api_key: System.get_env("OPENAI_API_KEY") ||
    raise("OPENAI_API_KEY is not set")

A quick helper module keeps the key lookup in one place:

defmodule MyApp.OpenAI do
  @endpoint "https://api.openai.com/v1/chat/completions"

  defp api_key, do: Application.fetch_env!(:my_app, :openai)[:api_key]
end

Calling the streaming endpoint

OpenAI’s chat completions endpoint streams when you pass stream: true. The response body arrives as server-sent events — a sequence of data: {...} lines, each carrying one delta, terminated by a final data: [DONE].

Req lets you handle the body incrementally with the :into option set to a function. That function is called with each raw chunk as it arrives off the socket. We parse the SSE lines, pull out the token, and forward it to whichever process asked for the completion:

defmodule MyApp.OpenAI do
  @endpoint "https://api.openai.com/v1/chat/completions"

  @doc """
  Streams a chat completion. Sends `{:chunk, token}` messages to `pid`
  for every token, then `{:done, reason}` when the stream ends.
  """
  def stream_completion(messages, pid) do
    body = %{
      model: "gpt-4o-mini",
      stream: true,
      messages: messages
    }

    Req.post(@endpoint,
      json: body,
      auth: {:bearer, api_key()},
      receive_timeout: 60_000,
      into: fn {:data, data}, acc ->
        for line <- parse_sse(data), do: dispatch(line, pid)
        {:cont, acc}
      end
    )
    |> case do
      {:ok, _resp} -> send(pid, {:done, :ok})
      {:error, err} -> send(pid, {:done, {:error, err}})
    end
  end

  defp api_key, do: Application.fetch_env!(:my_app, :openai)[:api_key]

  # Split a raw chunk into individual `data:` payloads.
  defp parse_sse(data) do
    data
    |> String.split("\n")
    |> Enum.filter(&String.starts_with?(&1, "data: "))
    |> Enum.map(&String.replace_prefix(&1, "data: ", ""))
  end

  defp dispatch("[DONE]", _pid), do: :ok

  defp dispatch(json, pid) do
    case Jason.decode(json) do
      {:ok, %{"choices" => [%{"delta" => %{"content" => token}} | _]}}
      when is_binary(token) ->
        send(pid, {:chunk, token})

      _ ->
        # keep-alive lines, role deltas, and finish markers land here
        :ok
    end
  end
end

Two things worth calling out. First, a single TCP chunk can contain a partial SSE line or several complete ones — filtering for lines that start with data: handles the common cases cleanly, and you can add a small buffer if you ever see a token split across chunks. Second, the very first delta contains a role field and no content, and OpenAI sends periodic keep-alive lines; the catch-all clause in dispatch/2 silently ignores both.

Wiring it into a LiveView

Now the fun part. When the user submits a prompt, we spawn the OpenAI call as an async task so the LiveView stays responsive, and we append each incoming token to the assistant message.

defmodule MyAppWeb.ChatLive do
  use MyAppWeb, :live_view

  @impl true
  def mount(_params, _session, socket) do
    {:ok,
     assign(socket,
       messages: [],
       pending: nil,
       streaming?: false
     )}
  end

  @impl true
  def handle_event("send", %{"prompt" => prompt}, socket) do
    history =
      socket.assigns.messages ++ [%{role: "user", content: prompt}]

    live_view = self()

    task =
      Task.Supervisor.async_nolink(MyApp.TaskSupervisor, fn ->
        MyApp.OpenAI.stream_completion(
          Enum.map(history, &Map.take(&1, [:role, :content])),
          live_view
        )
      end)

    {:noreply,
     socket
     |> assign(messages: history, pending: "", streaming?: true)
     |> assign(task_ref: task.ref)}
  end

We use Task.Supervisor.async_nolink/3 so a crash in the OpenAI call doesn’t take the LiveView down with it — you get a clean :DOWN message instead. Add the supervisor to your application.ex:

children = [
  # ...
  {Task.Supervisor, name: MyApp.TaskSupervisor}
]

Because stream_completion/2 sends {:chunk, token} and {:done, reason} messages back to the LiveView pid, we handle them in handle_info/2:

  @impl true
  def handle_info({:chunk, token}, socket) do
    {:noreply, update(socket, :pending, &(&1 <> token))}
  end

  def handle_info({:done, :ok}, socket) do
    finished =
      socket.assigns.messages ++
        [%{role: "assistant", content: socket.assigns.pending}]

    {:noreply,
     assign(socket, messages: finished, pending: nil, streaming?: false)}
  end

  def handle_info({:done, {:error, _reason}}, socket) do
    {:noreply,
     socket
     |> put_flash(:error, "The AI request failed. Please try again.")
     |> assign(pending: nil, streaming?: false)}
  end

  # async_nolink sends this when the task finishes or crashes.
  def handle_info({ref, _result}, socket) when is_reference(ref) do
    Process.demonitor(ref, [:flush])
    {:noreply, socket}
  end

  def handle_info({:DOWN, _ref, :process, _pid, _reason}, socket) do
    {:noreply, assign(socket, streaming?: false)}
  end

Every {:chunk, token} appends one token to :pending and triggers a re-render — LiveView sends only the diff of the changed text node, so this stays cheap even at hundreds of tokens.

Rendering the stream

The template renders the settled conversation plus the in-progress pending message. A blinking caret while streaming? is true sells the effect:

  @impl true
  def render(assigns) do
    ~H"""
    <div id="chat" class="space-y-4">
      <div :for={msg <- @messages} class={"message #{msg.role}"}>
        {msg.content}
      </div>

      <div :if={@pending} class="message assistant">
        {@pending}<span :if={@streaming?} class="caret">▋</span>
      </div>
    </div>

    <form phx-submit="send">
      <input type="text" name="prompt" placeholder="Ask anything…"
             disabled={@streaming?} autocomplete="off" />
      <button type="submit" disabled={@streaming?}>Send</button>
    </form>
    """
  end
end

That’s a complete, working streaming chat. The user types a prompt, tokens flow in live, and the input re-enables when the stream finishes. Because all the state is server-side assigns, a page reload or reconnect leaves you with a consistent conversation — no client cache to invalidate.

If you want the message list to scale to thousands of messages without re-diffing the whole history on every keystroke, move @messages into a LiveView stream and keep only the actively-streaming @pending string in assigns. Streams keep the settled history out of the socket’s memory while the token-by-token part stays in assigns where the frequent diffs happen.

Handling cancellation and real-world failures

Real users close tabs mid-answer, ask a follow-up before the first finishes, and hit network blips. A few things make this production-ready:

• Cancel on new input. Before spawning a new task, shut down the old one with Task.Supervisor.terminate_child/2 (or track the ref and ignore late chunks) so a stale stream can’t bleed tokens into the new answer.
• Timeouts. The receive_timeout: 60_000 on Req guards against a hung connection. Surface it as the same friendly error flash.
• Rate limits and 429s. When OpenAI pushes back, Req.post/2 returns a non-2xx response, not an exception. Match on the status and back off — a retry with jitter is usually enough.
• Process teardown is automatic. If the user navigates away, the LiveView process terminates. Because we used async_nolink under a Task.Supervisor, the orphaned task is cleaned up rather than leaking. This is the Elixir superpower: the lifecycle of your streaming work is tied to the lifecycle of the page, for free.

From tutorial to production

You now have the core loop — call OpenAI with stream: true, parse SSE with Req‘s :into callback, forward tokens as messages, and append them to a LiveView assign. That’s genuinely all streaming AI needs on the Elixir side.

Going to production, you’ll want the surrounding scaffolding too: authentication so only your users can burn tokens, per-user rate limiting, persistence of conversations, usage tracking for billing, and a component library so the chat UI looks finished. If you’d rather not rebuild all of that from scratch, our ai_template ships a Phoenix + LiveView AI starter with streaming chat, auth, and a polished interface already wired together — a running head start on exactly the pattern above.

If you’re building something more specialized, it’s worth seeing how the same streaming primitives compose in a real product. Our Introducing pdfai post walks through a document-analysis app built on LiveView and OpenAI, and MCP Chat Buddy shows the same ideas applied to chatting with a Postgres database through the Model Context Protocol. Both lean on the “stream tokens into assigns” pattern you just built.

PhxTemplates is the AI-native, use-case-specific Phoenix starter — at a lifetime price. If you’re shipping more than one idea, the Builder Pass gives you every template, so you can start each new AI project with the streaming, auth, and billing already done and spend your time on the part that’s actually yours.

Streaming used to be the hard, fiddly part of an AI feature. In Phoenix LiveView, it’s a Task, a send/2, and an update/3. Ship it.