# How OmniRoute Handles SSE Stream Responses: Architecture and Implementation

> Learn how OmniRoute handles SSE stream responses via a three-layer pipeline. Discover its architecture and implementation for event processing, back-pressure, and client-specific format transformations.

- Repository: [Diego Rodrigues de Sa e Souza/OmniRoute](https://github.com/diegosouzapw/OmniRoute)
- Tags: architecture
- Published: 2026-08-23

---

**OmniRoute handles SSE stream responses through a three-layer pipeline that reads upstream provider events, applies back-pressure and idle-timeout protection, normalizes tool-call data, and transforms the stream into client-specific formats like OpenAI Chat Completions or the Responses API.**

OmniRoute is an open-source LLM proxy that unifies access to 291+ providers through a single API. Understanding how OmniRoute handles SSE stream responses reveals the architecture behind its real-time streaming capabilities, from raw provider data ingestion to format-specific client delivery.

## The Three-Layer Streaming Architecture

OmniRoute’s streaming pipeline is organized into distinct layers that separate concerns between request handling, stream orchestration, and format transformation.

### Request Entry Point

The API routes (`/v1/chat/completions`, `/v1/responses`, and others) validate the incoming payload, select appropriate credentials, and forward the request to core handlers. According to the OmniRoute source code, these routes live in [`src/app/api/v1/.../route.ts`](https://github.com/diegosouzapw/OmniRoute/blob/main/src/app/api/v1/.../route.ts) and delegate to either [`open-sse/handlers/chatCore.ts`](https://github.com/diegosouzapw/OmniRoute/blob/main/open-sse/handlers/chatCore.ts) or [`open-sse/handlers/responsesHandler.ts`](https://github.com/diegosouzapw/OmniRoute/blob/main/open-sse/handlers/responsesHandler.ts) depending on the endpoint.

### SSE Stream Orchestration

This layer manages the connection to upstream providers. The `createSSEStream()` function in [`open-sse/utils/stream.ts`](https://github.com/diegosouzapw/OmniRoute/blob/main/open-sse/utils/stream.ts) reads the provider’s SSE, applies back-pressure handling, enforces idle-timeout protection, and normalizes tool-call data. It also estimates token usage and determines whether to emit the `[DONE]` terminator required by certain client specifications.

### API-Specific Transformation

The final layer converts the unified internal stream into the exact SSE schema expected by the client. For the Responses API, `createResponsesApiTransformStream()` in [`open-sse/transformer/responsesTransformer.ts`](https://github.com/diegosouzapw/OmniRoute/blob/main/open-sse/transformer/responsesTransformer.ts) handles this conversion, injecting keep-alive heartbeats and restructuring events to match OpenAI’s Responses API format.

## Core Streaming Logic in `createSSEStream`

The `createSSEStream()` function serves as the central orchestrator for all streaming operations in OmniRoute. It accepts a `StreamOptions` object that configures behavior across 291 supported providers.

```typescript
export function createSSEStream(options: StreamOptions = {}) {
  const {
    mode = STREAM_MODE.TRANSLATE,          // "translate" or "passthrough"
    targetFormat, sourceFormat,
    provider, model, connectionId,
    body, onComplete, onFailure,
    dropResponsesCommentary,
    // ...
  } = options;

```

### Idle Timeout and Back-Pressure Protection

OmniRoute prevents hung requests through an idle-timeout mechanism. As implemented in [`open-sse/utils/stream.ts`](https://github.com/diegosouzapw/OmniRoute/blob/main/open-sse/utils/stream.ts), a timer starts when the first data chunk arrives. If no subsequent chunks arrive within `STREAM_IDLE_TIMEOUT_MS`, the stream closes with an error:

```typescript
// From open-sse/utils/stream.ts#L997-L1002
const idleTimeout = setTimeout(() => {
  controller.error(new Error('Stream idle timeout exceeded'));
  cleanup();
}, STREAM_IDLE_TIMEOUT_MS);

```

The implementation also records the JSON byte length of request bodies as a performance mark (`omni-request-body-size`) for monitoring purposes, enabling detailed observability of streaming payloads.

### Pass-Through vs. Translate Modes

OmniRoute supports two operational modes for SSE handling:

- **Passthrough mode**: Forwards the upstream stream almost verbatim while still normalizing tool-call IDs, aggregating fragmented tool-call JSON fragments, and estimating token usage. Use this when the client and provider speak the same dialect.

- **Translate mode**: Parses the stream into a `TranslateState` object that enables emission of unified events like `think`, `output_text`, and `tool_calls`. This mode is essential when converting between provider-specific formats and standard client expectations.

### Usage Tracking and Termination Logic

When the upstream stream finishes, `createSSEStream` invokes `onComplete({status, usage})` to record per-model usage statistics. Errors route through `onFailure` for consistent error reporting across the proxy.

The function also determines whether to emit the `[DONE]` sentinel. OpenAI Chat Completions require this terminator, while the Responses API, Claude, and Antigravity providers terminate on their native final events. The boolean `shouldEmitDoneTerminator` drives emission logic at lines 888-894 of [`open-sse/utils/stream.ts`](https://github.com/diegosouzapw/OmniRoute/blob/main/open-sse/utils/stream.ts).

## Transforming for the Responses API

For the `/v1/responses` endpoint, OmniRoute first processes the request through `handleChatCore`, then rewires the SSE stream using `createResponsesApiTransformStream`.

```typescript
const transformStream = createResponsesApiTransformStream(null, undefined, {
  customToolNames,
});
const transformedBody = response.body
  .pipeThrough(transformStream)
  .pipeThrough(createSseHeartbeatTransform({
    signal,
    intervalMs: SSE_HEARTBEAT_INTERVAL_MS,
    shape: HEARTBEAT_SHAPES.OPENAI_RESPONSES_IN_PROGRESS,
  }));

```

### State Machine and Event Buffering

`createResponsesApiTransformStream` maintains an internal state machine that:

- Assigns deterministic `output_index` values to stream chunks
- Buffers partial text deltas and emits `response.output_text.delta` and `response.output_text.done` events
- Builds a dense, ordered `response.output` array once the stream ends
- Inserts reasoning items (`response.reasoning_summary_text.*`) when the client expects "thinking" content

### Keep-Alive Heartbeats

To prevent connection timeouts during slow generation, the transformer emits `event: ping` frames when no data arrives for `keepaliveIntervalMs` (defaulting to 3 seconds). This ensures that load balancers and client libraries maintain the connection throughout long-running inference tasks.

## Implementation Examples

### Direct SSE Stream Usage

For custom integrations, you can invoke `createSSEStream` directly in passthrough mode:

```typescript
import { createSSEStream } from "@/open-sse/utils/stream.ts";

export async function myStreamingHandler(req: Request) {
  const upstreamResponse = await fetch("https://api.provider.com/v1/stream", {
    method: "POST",
    body: req.body,
    headers: { "Accept": "text/event-stream" },
  });

  // Pipe the upstream SSE through OmniRoute's passthrough stream
  const transform = createSSEStream({ 
    mode: "passthrough", 
    provider: "my-provider" 
  });
  const streamed = upstreamResponse.body!.pipeThrough(transform);

  return new Response(streamed, {
    status: 200,
    headers: { "Content-Type": "text/event-stream" },
  });
}

```

In this configuration, `mode: "passthrough"` forwards events unchanged while still normalizing tool-call IDs and estimating usage for analytics.

### Handling the Responses API

For automatic conversion to the Responses API format, use the dedicated handler:

```typescript
import { handleResponsesCore } from "@/open-sse/handlers/responsesHandler.ts";

export async function responsesRoute(req: Request) {
  const body = await req.json();
  const result = await handleResponsesCore({
    body,
    modelInfo: { provider: "openai", model: "gpt-4o-mini" },
    credentials: { apiKey: process.env.OPENAI_API_KEY },
    connectionId: crypto.randomUUID(),
    signal: req.signal,
  });

  // `result` is either a Response (already streaming) or an error envelope
  if (result instanceof Response) return result;
  return new Response(
    JSON.stringify({ error: result.error }), 
    { status: result.status }
  );
}

```

This handler internally converts the Responses payload to an OpenAI-compatible chat request, invokes the generic chat core, then re-pipes the upstream SSE through `createResponsesApiTransformStream` to emit the Responses API schema.

## Summary

- OmniRoute processes SSE streams through three layers: request entry, stream orchestration, and API-specific transformation.
- The `createSSEStream()` function in [`open-sse/utils/stream.ts`](https://github.com/diegosouzapw/OmniRoute/blob/main/open-sse/utils/stream.ts) manages idle timeouts, back-pressure, and tool-call normalization across 291 providers.
- **Passthrough mode** preserves the provider format while adding metadata; **translate mode** converts between dialects using an internal state machine.
- The Responses API uses `createResponsesApiTransformStream()` to reshape Chat Completions into the Responses schema with keep-alive heartbeats.
- Usage tracking and completion callbacks enable per-model analytics and robust error handling throughout the streaming lifecycle.

## Frequently Asked Questions

### What is the difference between passthrough and translate modes in OmniRoute?

**Passthrough mode** forwards provider SSE events with minimal modification, primarily normalizing tool-call IDs and estimating token usage. **Translate mode** parses the stream into a unified internal representation, enabling conversion between provider-specific formats and standard client expectations like OpenAI Chat Completions or Claude's streaming protocol.

### How does OmniRoute prevent streaming requests from hanging indefinitely?

OmniRoute implements an **idle-timeout mechanism** in `createSSEStream()` that starts when the first data chunk arrives. If no subsequent chunks arrive within `STREAM_IDLE_TIMEOUT_MS`, the stream automatically closes with an error, preventing resource exhaustion from stalled connections.

### Does OmniRoute support tool calls in SSE streams?

Yes. OmniRoute normalizes tool-call data across all streaming modes. In `createSSEStream()`, the implementation aggregates fragmented tool-call JSON fragments, assigns deterministic IDs, and ensures proper formatting regardless of whether the upstream provider sends complete or partial tool-call objects.

### How does OmniRoute transform Chat Completions into the Responses API format?

The `handleResponsesCore` function in [`open-sse/handlers/responsesHandler.ts`](https://github.com/diegosouzapw/OmniRoute/blob/main/open-sse/handlers/responsesHandler.ts) first converts the request to a standard chat format, then pipes the response through `createResponsesApiTransformStream()`. This transformer maintains a state machine that buffers text deltas, assigns `output_index` values, and emits the `response.output_text.delta` events required by the Responses API specification.