How OmniRoute Handles SSE Stream Responses: Architecture and Implementation

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 and delegate to either open-sse/handlers/chatCore.ts or 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 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 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.

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, 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:

// 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.

Transforming for the Responses API

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

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:

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:

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 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 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.

{
  "mcpServers": {
    "instagit": {
      "command": "npx",
      "args": ["-y", "instagit@latest"]
    }
  }
}