How to Customize Streaming Behavior in OmniRoute: 5 Methods for Full Control

You can customize streaming behavior in OmniRoute at five distinct levels: per-request via headers or JSON flags, per-API-key via database defaults, through custom SSE payload transformers, by adjusting telemetry limits, and using feature-flag middleware overrides.

OmniRoute provides a modular streaming pipeline that gives you granular control over how Server-Sent Events (SSE) are generated, delivered, and monitored. Whether you need to force streaming for specific clients, change default behaviors per API key, or inject custom transformations into the SSE payload, the codebase exposes explicit hooks to customize streaming behavior in OmniRoute without forking the core.

Per-Request Streaming Controls

The entry point at src/sse/handlers/chat.ts evaluates every request to determine if it should stream based on the Accept header and the request body.

Forcing Streaming with the Accept Header

When the Accept: text/event-stream header is present, the acceptHeaderForcesStream function automatically sets stream: true on the upstream request. This is the standard HTTP-level toggle that overrides any implicit defaults.

curl -H "Accept: text/event-stream" \
     -H "Authorization: Bearer $OMNIRoute_API_KEY" \
     -d '{"model":"gpt-4","messages":[{"role":"user","content":"Explain streaming"}]}' \
     https://api.omniroute.com/v1/chat/completions

Disabling Streaming via Request Body

Conversely, setting "stream": false in the JSON payload overrides the header and forces a non-streaming response. This gives clients explicit control regardless of headers.

curl -H "Authorization: Bearer $OMNIRoute_API_KEY" \
     -d '{"model":"gpt-4","messages":[{"role":"user","content":"Hello"}],"stream":false}' \
     https://api.omniroute.com/v1/chat/completions

Per-API-Key Default Modes

When clients omit the stream flag, OmniRoute falls back to the stream_default_mode column defined in migration src/lib/db/migrations/077_api_key_stream_default_mode.sql. The accessor in src/lib/db/apiKeys.ts reads this value and injects it into the request context before routing.

The column supports values like legacy or default, allowing you to migrate clients gradually without changing their code.

UPDATE api_keys
SET stream_default_mode = 'default'
WHERE key = 'abcd1234';

You can also update this via the admin API if exposed:

curl -X PATCH https://api.omniroute.com/v1/keys/abcd1234 \
     -H "Authorization: Bearer $ADMIN_TOKEN" \
     -d '{"stream_default_mode":"default"}'

Custom SSE Payload Transformation

For reshaping SSE chunks—such as converting OpenAI-format events to the Responses API format—OmniRoute uses src/lib/translator/streamTransform.ts. You can chain custom TransformStreams before or after the built-in createResponsesApiTransformStream to modify payloads without altering the upstream provider.

import { createResponsesApiTransformStream } from "@omniroute/open-sse/transformer/responsesTransformer.ts";

export async function transformWithTimestamp(rawSse: string): Promise<string> {
  const encoder = new TextEncoder();
  const decoder = new TextDecoder();

  const input = new ReadableStream({
    start(controller) {
      controller.enqueue(encoder.encode(rawSse));
      controller.close();
    },
  });

  const timestampTransform = new TransformStream({
    transform(chunk, ctrl) {
      const text = decoder.decode(chunk);
      ctrl.enqueue(encoder.encode(`${text}\ntimestamp:${Date.now()}`));
    },
  });

  const output = input
    .pipeThrough(timestampTransform)
    .pipeThrough(createResponsesApiTransformStream());

  const reader = output.getReader();
  let result = "";
  while (true) {
    const { value, done } = await reader.read();
    if (done) break;
    result += decoder.decode(value, { stream: true });
  }
  result += decoder.decode(); // flush
  return result;
}

Stream Telemetry and Lifecycle Limits

The StreamTracker class in src/sse/services/streamState.ts maintains a state machine tracking transitions from INITIALIZED through STREAMING to completion. By default, it retains the last 50 completed streams in memory, configurable via the STREAM_HISTORY_MAX environment variable.

You can expose this telemetry via your own endpoint:

import { getActiveStreams } from "@omniroute/open-sse/services/streamState";

export async function streamMetricsHandler(req, res) {
  res.json({ activeStreams: getActiveStreams() });
}

Set STREAM_HISTORY_MAX=200 to increase the buffer without code changes.

Feature-Flag Overrides for Testing

As demonstrated in tests/unit/resolve-stream-flag.test.ts, you can inject middleware that mutates req.body.stream based on custom headers like X-OmniRoute-Force-Stream. This enables runtime toggles for A/B testing or emergency overrides without client changes.

export function forceStreamMiddleware(req, res, next) {
  if (req.headers["x-omniroute-force-stream"] === "true") {
    req.body.stream = true;
  } else if (req.headers["x-omniroute-force-stream"] === "false") {
    req.body.stream = false;
  }
  next();
}

Add this middleware before the chat handler in your route stack to intercept requests dynamically.

Summary

• Per-request toggles: Use the Accept: text/event-stream header or stream field in the request body to control streaming behavior in src/sse/handlers/chat.ts.
• API-key defaults: Set the stream_default_mode column in the api_keys table (added in migration 077) to define defaults via src/lib/db/apiKeys.ts.
• Payload transformation: Chain custom TransformStreams in src/lib/translator/streamTransform.ts to reshape SSE chunks before they reach the client.
• Telemetry limits: Adjust the STREAM_HISTORY_MAX environment variable to control how many completed streams the StreamTracker retains in src/sse/services/streamState.ts.
• Runtime overrides: Implement middleware to force stream values based on custom headers for testing or operational flexibility.

Frequently Asked Questions

How does OmniRoute decide whether to stream when the client doesn't specify?

When the client omits the stream flag, OmniRoute queries the stream_default_mode column from the api_keys table via src/lib/db/apiKeys.ts. This per-key default—set to either legacy or default—determines the streaming behavior before the request reaches the upstream handler.

Can I modify the SSE payload format without changing the upstream provider?

Yes. The createResponsesApiTransformStream function in src/lib/translator/streamTransform.ts exposes a standard Web Streams API interface. You can pipe the raw SSE through your own TransformStream to inject metadata, filter events, or convert formats before the final response reaches the client.

What is the maximum number of streams OmniRoute tracks for telemetry?

By default, the StreamTracker class in src/sse/services/streamState.ts retains the last 50 completed streams. You can increase this limit by setting the STREAM_HISTORY_MAX environment variable to any integer value, adjusting the history buffer without code changes.

Can I force streaming on or off for specific requests without modifying client code?

Yes. You can add middleware that checks for custom headers like X-OmniRoute-Force-Stream and mutates req.body.stream before the request hits src/sse/handlers/chat.ts. This pattern, illustrated in tests/unit/resolve-stream-flag.test.ts, allows runtime overrides via API gateway rules or reverse proxy configurations.

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