@nath/sse-client

Lightweight SSE (Server-Sent Events) client for Node.js & browsers. Handles single streams, parallel streams, streaming text merging, and custom extractors — no dependencies.

Install

npm install @nath/sse-client

Basic Usage

Single stream — just listen to raw events

import { createSseParser } from "@nath/sse-client"

await createSseParser("https://example.com/stream", {
  onEvent: (event) => {
    console.log(event.event, event.data)
  }
})

Merge streaming text

Got an AI API that streams token by token? Use eventMerge to accumulate the full text as it comes in.

await createSseParser("https://example.com/stream", {
  eventMerge: true,

  onMessage: (full) => {
    process.stdout.write("\r" + full)
  },

  fetchOptions: {
    method: "POST",
    headers: {
      Authorization: "Bearer YOUR_KEY",
      "Content-Type": "application/json"
    },
    body: JSON.stringify({
      model: "gpt-4o",
      stream: true,
      messages: [{ role: "user", content: "yo explain SSE" }]
    })
  }
})

The built-in extractor auto-detects these fields from each event's JSON: text → delta → content → choices[0].delta.content

Custom extractor — non-standard response format

If the API returns something weird, just tell it how to extract the chunk:

await createSseParser("https://example.com/stream", {
  eventMerge: true,

  extractor: (json) => {
    return json?.output?.text ?? null
  },

  onMessage: (full) => {
    console.log(full)
  }
})

Return null to skip an event (e.g. progress events, non-text events).

Parallel streams — Hydra mode

Need to hit multiple SSE endpoints at the same time and handle them with the same logic? That's what createSseHydra is for.

import { createSseHydra } from "@nath/sse-client"

await createSseHydra(
  [
    { url: "https://example.com/stream-1" },
    { url: "https://example.com/stream-2" },
    { url: "https://example.com/stream-3" }
  ],
  {
    eventMerge: true,

    onMessage: (full, source) => {
      console.log(`[${source}]`, full)
    }
  }
)

Each stream runs in parallel. The source param tells you which URL the message came from — useful when responses arrive interleaved.

Per-stream fetch options (auth, body, headers)

Each input in Hydra can have its own fetchOptions:

await createSseHydra(
  [
    {
      url: "https://example.com/stream-1",
      fetchOptions: {
        method: "POST",
        headers: { Authorization: "Bearer TOKEN_1" },
        body: JSON.stringify({ query: "hello" })
      }
    },
    {
      url: "https://example.com/stream-2",
      fetchOptions: {
        method: "POST",
        headers: { Authorization: "Bearer TOKEN_2" },
        body: JSON.stringify({ query: "world" })
      }
    }
  ],
  {
    onEvent: (event, source) => {
      console.log(source, event)
    }
  }
)

Timeout

Default timeout is 120000ms (2 minutes). Override it:

await createSseParser("https://example.com/stream", {
  timeout: 30000,
  onEvent: (event) => console.log(event)
})

Stream gets aborted cleanly when timeout hits — no crash, no unhandled rejection.

API

createSseParser(url, options?)

| Option | Type | Description | |---|---|---| | onEvent | (event, source?) => void | Called for every raw SSE event | | onMessage | (merged, source?) => void | Called with accumulated text (requires eventMerge: true) | | eventMerge | boolean | Enable text accumulation mode | | extractor | (json) => string \| null | Custom chunk extractor from parsed JSON | | timeout | number | Abort after N ms (default: 120000) | | fetchOptions | RequestInit | Passed directly to fetch() |

createSseHydra(inputs, options?)

Same options as above. inputs is an array of { url, fetchOptions? }.

SSEEvent shape

type SSEEvent = {
  id?: string       // event ID from server
  event?: string    // event name/type
  data: string      // raw data string
  retry?: number    // retry interval hint from server (ms)
}

Notes

• Works in Node.js 18+ and modern browsers (uses native fetch)
• No dependencies
• AbortError on timeout is swallowed silently — everything else is rethrown
• If you pass your own signal in fetchOptions, it's merged with the internal timeout controller