Implementation:Openai Openai node Stream

Knowledge Sources	Openai_Openai_node
Domains	SDK, Streaming
Last Updated	2026-02-15 12:00 GMT

Overview

The Stream class is the foundational generic wrapper for consuming Server-Sent Events (SSE) responses from the OpenAI API as async iterables.

Description

Stream<Item> implements AsyncIterable<Item>, allowing consumers to use for await...of loops to process streamed data. It is the core primitive upon which higher-level streaming abstractions such as ChatCompletionStream, AssistantStream, and ResponseStream are built.

The class provides two static factory methods for constructing streams: fromSSEResponse parses a raw HTTP Response with an SSE body by decoding chunks, splitting on double newlines, and yielding parsed JSON objects (stopping when the [DONE] sentinel is received). fromReadableStream takes a newline-delimited ReadableStream of JSON values, typically one produced by toReadableStream(), and yields each parsed line as an item.

The stream tracks an AbortController that is used to cancel the underlying HTTP request. If the consumer breaks out of the iteration early, the controller is automatically aborted. The tee() method splits a stream into two independent streams that can be consumed at different speeds, and toReadableStream() serializes the stream back into a newline-delimited ReadableStream of JSON strings for cross-boundary transport (e.g., server to browser).

Usage

This class is used internally whenever a streaming API call is made (e.g., client.chat.completions.create({ stream: true })). It is also used directly when bridging streams between server and client via Stream.fromReadableStream() and .toReadableStream().

Code Reference

Source Location

Repository: openai-node
File: src/core/streaming.ts
Lines: 1-348

Signature

export class Stream<Item> implements AsyncIterable<Item> {
  controller: AbortController;

  constructor(
    iterator: () => AsyncIterator<Item>,
    controller: AbortController,
    client?: OpenAI,
  );

  static fromSSEResponse<Item>(
    response: Response,
    controller: AbortController,
    client?: OpenAI,
  ): Stream<Item>;

  static fromReadableStream<Item>(
    readableStream: ReadableStream,
    controller: AbortController,
    client?: OpenAI,
  ): Stream<Item>;

  [Symbol.asyncIterator](): AsyncIterator<Item>;
  tee(): [Stream<Item>, Stream<Item>];
  toReadableStream(): ReadableStream;
}

Import

import { Stream } from 'openai/streaming';

I/O Contract

Inputs

Name	Type	Required	Description
iterator	`() => AsyncIterator<Item>`	Yes	Factory function that returns the async iterator producing stream items
controller	`AbortController`	Yes	Controller used to abort the underlying HTTP request
client	`OpenAI`	No	Client instance used for logging

Outputs

Name	Type	Description
Stream<Item>	`AsyncIterable<Item>`	An async iterable that yields parsed items from the SSE stream
tee()	`[Stream<Item>, Stream<Item>]`	Two independent streams reading from the same source
toReadableStream()	`ReadableStream`	A web-standard ReadableStream of newline-delimited JSON

Usage Examples

Basic Usage

import OpenAI from 'openai';

const client = new OpenAI();
const stream = await client.chat.completions.create({
  model: 'gpt-4o',
  messages: [{ role: 'user', content: 'Hello' }],
  stream: true,
});

for await (const chunk of stream) {
  process.stdout.write(chunk.choices[0]?.delta?.content || '');
}

Splitting a Stream

const [stream1, stream2] = stream.tee();

// Consume independently
for await (const chunk of stream1) { /* logging */ }
for await (const chunk of stream2) { /* processing */ }

Server-to-Client Transport

// Server side: convert to ReadableStream for HTTP response
const readableStream = stream.toReadableStream();

// Client side: reconstruct from ReadableStream
const clientStream = Stream.fromReadableStream<ChatCompletionChunk>(readableStream, new AbortController());
for await (const chunk of clientStream) {
  // process chunk
}

Related Pages

Page Connections

Double-click a node to navigate. Hold to expand connections.

Principle

Implementation

Heuristic

Environment