Examples of Using OmniRoute's Streaming Engine: Implementation Guide and Code Samples

OmniRoute's streaming engine implements a full-stack Server-Sent Events (SSE) pipeline that wraps upstream LLM responses into disconnect-aware streams, handling back-pressure, error conversion, and graceful termination through utilities like createDisconnectAwareStream and finalizeReadableStream.

The diegosouzapw/OmniRoute repository provides a Next.js-based routing layer for LLM APIs that supports real-time streaming via Server-Sent Events. This guide provides concrete examples of using OmniRoute's streaming engine, from low-level stream utilities to CLI commands, based on the actual source code implementation in the project.

Core Components of the Streaming Pipeline

OmniRoute's streaming architecture processes requests through a coordinated pipeline of handlers and utilities that transform provider responses into compliant SSE streams.

API Entry Points and Route Handlers

Streaming requests enter through Next.js API routes. In src/app/api/v1/relay/chat/completions/route.ts, requests are received and passed to the stream finalization layer. The route handler acts as the public interface, accepting client connections and preparing the SSE response headers.

The Stream Finalization Layer

The src/app/api/v1/relay/chat/completions/streamFinalizer.ts module wraps upstream responses in a ReadableStream and attaches a finalizer that emits the [DONE] marker. This component handles abort signals and disconnect logic, ensuring that every stream terminates cleanly regardless of upstream errors.

Core Chat Handler and Routing

In open-sse/handlers/chatCore.ts, the pipeline forwards requests to the combo router, which dispatches to provider-specific executors. This handler coordinates between the incoming request format and the appropriate backend streaming implementation.

Streaming Utilities and Helper Functions

The open-sse/utils/ directory contains the low-level engine components that manage stream lifecycle events.

Disconnect-Aware Stream Wrappers

The createDisconnectAwareStream function in open-sse/utils/streamHandler.ts converts upstream errors or client disconnects into SSE-compatible error chunks. It guarantees that a [DONE] termination marker is always sent, even when the upstream connection fails.

The companion createStreamController utility (same file) supplies callback hooks including onError and onDisconnect that enable telemetry collection and resource cleanup.

Stream Readiness and Payload Collection

open-sse/utils/streamReadinessPolicy.ts implements shouldStream logic that determines whether a request should be streamed based on provider capabilities and client flags. Meanwhile, open-sse/utils/streamPayloadCollector.ts buffers partial chunks, updates usage statistics, and forwards aggregated payloads downstream.

Code Examples: Implementing OmniRoute Streaming

These examples demonstrate how to use OmniRoute's streaming engine in different contexts, derived from the test suite and source implementations.

Example 1: Wrapping an Upstream ReadableStream for SSE Output

This pattern from tests/unit/stream-handler.test.ts shows how to wrap a provider response using the core streaming utilities:

import { createDisconnectAwareStream, createStreamController } from
  '../../open-sse/utils/streamHandler.ts';
import { finalizeReadableStream } from
  '../../src/app/api/v1/relay/chat/completions/streamFinalizer.ts';

// Mock upstream transform stream (e.g., the result of a provider executor)
const upstream = {
  readable: new ReadableStream({
    start(ctrl) {
      // Simulate a normal streaming response
      ctrl.enqueue(new TextEncoder().encode('data: {"choices":[{"delta":{"content":"Hello"}}]}\n\n'));
      // End of stream
      ctrl.enqueue(new TextEncoder().encode('data: [DONE]\n\n'));
      ctrl.close();
    },
  }),
  writable: { getWriter: () => ({ abort() {} }) },
};

// Attach disconnect-aware wrapper
const sseStream = createDisconnectAwareStream(
  upstream,
  createStreamController({
    // Optional hooks for logging/metrics
    onError: (err) => console.error('SSE error:', err),
    onDisconnect: () => console.warn('Client disconnected')
  })
);

// The finalizer adds the finishing event and makes the stream compatible with Next.js
export const response = new Response(finalizeReadableStream(sseStream), {
  headers: { 'Content-Type': 'text/event-stream; charset=utf-8' },
});

Example 2: Streaming from the Command Line

OmniRoute exposes streaming functionality through its CLI. As documented in skills/cli-chat/SKILL.md, you can invoke the streaming engine directly:

# Run the built-in CLI command that streams a prompt to the default provider

omniroute stream "Explain quantum entanglement"

Programmatically, you can spawn this process to consume the SSE stream:

import { spawn } from 'node:child_process';

const proc = spawn('omniroute', ['stream', 'Explain quantum entanglement']);
proc.stdout.on('data', (chunk) => {
  // Each chunk is an SSE line; strip the SSE prefix for pretty printing
  const line = chunk.toString().replace(/^data: /, '');
  console.log('🧩', line);
});
proc.stderr.on('data', (err) => console.error('❗', err.toString()));

Example 3: Server-Side Stream Readiness Checks

Before initiating a stream, check if the configuration supports it using the readiness policy:

import { shouldStream } from '../../open-sse/utils/streamReadinessPolicy.ts';
import { FORMATS } from '../../open-sse/translator/formats.ts';

function handleChat(req) {
  const format = req.headers['accept'] ?? FORMATS.OPENAI_CHAT; // default
  const canStream = shouldStream({ 
    format, 
    clientWantsStream: req.query.stream === 'true' 
  });
  
  if (canStream) {
    // Route through streaming pipeline using createDisconnectAwareStream
  } else {
    // Fall back to non-streaming JSON response
  }
}

Error Handling and Disconnection Management

The tests/unit/stream-handler.test.ts file demonstrates how the streaming engine handles edge cases:

• Basic disconnect handling: When a mock upstream stream errors, createDisconnectAwareStream ensures the client receives an SSE error object containing finish_reason:"error" followed by a [DONE] marker.
• Post-completion error suppression: Any error occurring after the [DONE] token is ignored, ensuring clients see a clean termination.
• Provider-specific error conversion: The test suite shows conversion of HTTP 503 errors (such as Gemini high-demand errors) into SSE error chunks while preserving the original error message in the payload.

These behaviors ensure that client applications receive predictable stream termination even when upstream providers fail.

Summary

• OmniRoute's streaming engine constructs SSE pipelines through src/app/api/v1/relay/chat/completions/streamFinalizer.ts and open-sse/handlers/chatCore.ts.
• Stream utilities in open-sse/utils/streamHandler.ts provide createDisconnectAwareStream for error conversion and createStreamController for lifecycle hooks.
• Client scenarios include wrapping upstream ReadableStreams, invoking streams via the omniroute stream CLI command, and checking stream readiness with shouldStream.
• Error resilience is guaranteed through automatic [DONE] marker injection and graceful handling of upstream disconnections.

Frequently Asked Questions

How does OmniRoute handle client disconnections during streaming?

When a client disconnects, the createDisconnectAwareStream utility in open-sse/utils/streamHandler.ts detects the broken connection through the abort signal. It invokes the onDisconnect callback for cleanup and ensures the stream terminates with a [DONE] marker, preventing resource leaks.

Can I use OmniRoute's streaming engine without the Next.js API routes?

Yes. While the repository provides Next.js route handlers in src/app/api/v1/relay/chat/completions/route.ts, the core streaming utilities are framework-agnostic. You can import createDisconnectAwareStream and finalizeReadableStream into any Node.js or edge runtime environment to build custom SSE endpoints.

What error format does OmniRoute use for streaming failures?

The engine converts upstream HTTP errors into SSE-compatible JSON chunks containing finish_reason:"error" and preserves the original message. For example, a 503 error from a provider becomes a data chunk with error details, followed by the standard [DONE] termination marker, as demonstrated in tests/unit/stream-handler.test.ts.

Where can I find additional runnable examples of the streaming utilities?

The tests/unit/stream-handler.test.ts file contains executable test cases that demonstrate disconnect handling, error conversion, and graceful completion. Additionally, src/lib/a2a/streaming.ts implements a JSON-RPC over SSE endpoint that shows production usage of the streaming pipeline for A2A (Agent-to-Agent) communication.

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