Streaming Responses

SSE Event Stream Format

The backend streams a series of text/event-stream events for each conversation turn:

// SSE stream from POST /api/chat/{sessionId}/send

// Event: conversation turn started
event: turn_start
data: {"turn_id":"tn-abc123","agent_id":"hr-assistant"}

// Event: tool call initiated (agent is running a tool)
event: tool_call
data: {"tool_name":"get_leave_balance","tool_call_id":"tc-xyz789"}

// Event: tool call completed
event: tool_result
data: {"tool_call_id":"tc-xyz789","elapsed_ms":245}

// Events: token-by-token LLM response
event: token
data: {"text":"You"}

event: token
data: {"text":" have"}

event: token
data: {"text":" 15"}

event: token
data: {"text":" annual"}

event: token
data: {"text":" leave"}

event: token
data: {"text":" days"}

// Event: conversation turn complete
event: turn_end
data: {"turn_id":"tn-abc123","total_tokens":42,"elapsed_ms":1850}

// Stream closed by server

Backend ISSEResponseWriter

// ISSEResponseWriter — abstracts SSE writing
public interface ISSEResponseWriter
{
    Task WriteEventAsync(string eventName, object data, CancellationToken ct);
    Task FlushAsync(CancellationToken ct);
    Task CloseAsync();
}

// Usage in the chat controller
public class ChatController : ControllerBase
{
    [HttpPost("{sessionId}/send")]
    public async Task SendMessageAsync(string sessionId,
        [FromBody] SendMessageRequest request, CancellationToken ct)
    {
        Response.ContentType = "text/event-stream";
        Response.Headers["Cache-Control"] = "no-cache";
        Response.Headers["X-Accel-Buffering"] = "no";  // Disable nginx buffering

        var writer = _sseWriterFactory.Create(Response);

        await writer.WriteEventAsync("turn_start", new { turn_id = Guid.NewGuid() }, ct);

        await foreach (var token in _agentEngine.StreamAsync(sessionId, request.Message, ct))
        {
            await writer.WriteEventAsync("token", new { text = token }, ct);
        }

        await writer.WriteEventAsync("turn_end", new { }, ct);
        await writer.CloseAsync();
    }
}

Frontend SSE Client

// TypeScript — widget SSE client
class OctopusSSEClient {
  private eventSource: EventSource | null = null;

  async sendMessage(sessionId: string, message: string, token: string) {
    const response = await fetch(`/api/chat/${sessionId}/send`, {
      method:  'POST',
      headers: {
        'Content-Type':  'application/json',
        'Authorization': `Bearer ${token}`
      },
      body: JSON.stringify({ message })
    });

    const reader  = response.body!.getReader();
    const decoder = new TextDecoder();
    let   buffer  = '';

    while (true) {
      const { done, value } = await reader.read();
      if (done) break;

      buffer += decoder.decode(value, { stream: true });
      const events = buffer.split('\n\n');
      buffer = events.pop() ?? '';

      for (const rawEvent of events) {
        const parsed = this.parseSSEEvent(rawEvent);
        if (parsed) this.handleEvent(parsed);
      }
    }
  }

  private handleEvent(event: { type: string; data: unknown }) {
    switch (event.type) {
      case 'token':
        this.onToken((event.data as any).text);
        break;
      case 'tool_call':
        this.onToolCall((event.data as any).tool_name);
        break;
      case 'turn_end':
        this.onComplete();
        break;
    }
  }
}

Typing Indicator States