mcpengine/studio/packages/ai-pipeline/streaming/sse.ts

// SSE Helper — converts async generator of PipelineEvents to Server-Sent Events Response

import type { PipelineEvent } from '../types';

/**
 * Create a Server-Sent Events Response from a PipelineEvent async generator.
 * Compatible with Next.js App Router API routes.
 */
export function createSSEResponse(
  generator: AsyncGenerator<PipelineEvent>
): Response {
  const encoder = new TextEncoder();

  const stream = new ReadableStream({
    async start(controller) {
      try {
        for await (const event of generator) {
          const data = `data: ${JSON.stringify(event)}\n\n`;
          controller.enqueue(encoder.encode(data));
        }
        // Send done event
        controller.enqueue(encoder.encode('data: [DONE]\n\n'));
      } catch (error) {
        const errorEvent: PipelineEvent = {
          type: 'error',
          message: error instanceof Error ? error.message : String(error),
          recoverable: false,
        };
        controller.enqueue(
          encoder.encode(`data: ${JSON.stringify(errorEvent)}\n\n`)
        );
      } finally {
        controller.close();
      }
    },
  });

  return new Response(stream, {
    headers: {
      'Content-Type': 'text/event-stream',
      'Cache-Control': 'no-cache, no-transform',
      Connection: 'keep-alive',
      'X-Accel-Buffering': 'no', // Disable nginx buffering
    },
  });
}

/**
 * Create a named SSE event (with event: field).
 * Useful when clients want to use EventSource.addEventListener().
 */
export function createNamedSSEResponse(
  generator: AsyncGenerator<PipelineEvent>
): Response {
  const encoder = new TextEncoder();

  const stream = new ReadableStream({
    async start(controller) {
      try {
        for await (const event of generator) {
          const eventType = event.type.replace(':', '_');
          const line = `event: ${eventType}\ndata: ${JSON.stringify(event)}\n\n`;
          controller.enqueue(encoder.encode(line));
        }
        controller.enqueue(encoder.encode('event: done\ndata: {}\n\n'));
      } catch (error) {
        const errorEvent: PipelineEvent = {
          type: 'error',
          message: error instanceof Error ? error.message : String(error),
          recoverable: false,
        };
        controller.enqueue(
          encoder.encode(`event: error\ndata: ${JSON.stringify(errorEvent)}\n\n`)
        );
      } finally {
        controller.close();
      }
    },
  });

  return new Response(stream, {
    headers: {
      'Content-Type': 'text/event-stream',
      'Cache-Control': 'no-cache, no-transform',
      Connection: 'keep-alive',
      'X-Accel-Buffering': 'no',
    },
  });
}