Streaming & SSE

Soxom supports Server-Sent Events (SSE) for real-time data streaming, ideal for AI/LLM applications and live updates.

Configuration

soxom.yaml

streaming:
  sse:
    enabled: true
    # Hide sentinel/termination events from SDK users
    sentinel_events:
      - "[DONE]"
      - "END"

Sentinel Events

Many streaming APIs use sentinel events to signal end of stream:

data: {"content": "Hello"}

data: {"content": " world"}

data: [DONE]

Configure sentinel_events to automatically filter these out.

OpenAPI Definition

Define streaming endpoints with text/event-stream content type:

paths:
  /chat/completions:
    post:
      operationId: createChatCompletion
      summary: Create a chat completion
      description: |
        When stream: true, returns Server-Sent Events.
      requestBody:
        required: true
        content:
          application/json:
            schema:
              $ref: '#/components/schemas/ChatCompletionRequest'
      responses:
        "200":
          description: Chat completion response
          content:
            # Non-streaming response
            application/json:
              schema:
                $ref: '#/components/schemas/ChatCompletion'
            # Streaming response
            text/event-stream:
              schema:
                $ref: '#/components/schemas/ChatCompletionChunk'

components:
  schemas:
    ChatCompletionRequest:
      type: object
      required:
        - model
        - messages
      properties:
        model:
          type: string
        messages:
          type: array
          items:
            $ref: '#/components/schemas/ChatMessage'
        stream:
          type: boolean
          default: false

    ChatCompletionChunk:
      type: object
      properties:
        id:
          type: string
        choices:
          type: array
          items:
            type: object
            properties:
              delta:
                type: object
                properties:
                  content:
                    type: string
              finish_reason:
                type: string
                nullable: true

Generated SDK Usage

Streaming Response

TypeScript

// Enable streaming with stream: true
const stream = await client.chat.completions.create({
  model: "gpt-4",
  messages: [
    { role: "user", content: "Write a haiku about coding" }
  ],
  stream: true
});

// Iterate over chunks
for await (const chunk of stream) {
  // Each chunk contains incremental content
  const content = chunk.choices[0]?.delta?.content;
  if (content) {
    process.stdout.write(content);
  }
}

Python

# Enable streaming with stream=True
stream = client.chat.completions.create(
    model="gpt-4",
    messages=[
        {"role": "user", "content": "Write a haiku about coding"}
    ],
    stream=True
)

# Iterate over chunks
for chunk in stream:
    content = chunk.choices[0].delta.content
    if content:
        print(content, end="", flush=True)

Go

// Enable streaming
stream, err := client.Chat.Completions.Create(ctx, ChatCompletionRequest{
    Model: "gpt-4",
    Messages: []ChatMessage{
        {Role: "user", Content: "Write a haiku about coding"},
    },
    Stream: true,
})
if err != nil {
    log.Fatal(err)
}
defer stream.Close()

// Iterate over chunks
for stream.Next() {
    chunk := stream.Current()
    if content := chunk.Choices[0].Delta.Content; content != "" {
        fmt.Print(content)
    }
}
if err := stream.Err(); err != nil {
    log.Fatal(err)
}

Non-Streaming Response

Same endpoint without streaming:

TypeScript

// Non-streaming (default)
const completion = await client.chat.completions.create({
  model: "gpt-4",
  messages: [
    { role: "user", content: "Write a haiku about coding" }
  ]
});

console.log(completion.choices[0].message.content);

Python

# Non-streaming (default)
completion = client.chat.completions.create(
    model="gpt-4",
    messages=[
        {"role": "user", "content": "Write a haiku about coding"}
    ]
)

print(completion.choices[0].message.content)

SSE Event Fields

Soxom supports all standard SSE fields:

Field	Description
id	Event identifier
event	Event type
data	Event payload (JSON parsed automatically)
retry	Reconnection time

Event Format

id: 1
event: message
data: {"content": "Hello"}

id: 2
event: message
data: {"content": " world"}

event: done
data: [DONE]

Handling Events

By Event Type

TypeScript

const stream = await client.events.subscribe();

for await (const event of stream) {
  switch (event.type) {
    case 'message':
      console.log('Message:', event.data);
      break;
    case 'error':
      console.error('Error:', event.data);
      break;
    case 'heartbeat':
      // Ignore heartbeats
      break;
  }
}

Python

stream = client.events.subscribe()

for event in stream:
    if event.type == "message":
        print("Message:", event.data)
    elif event.type == "error":
        print("Error:", event.data)

Collecting Full Response

TypeScript

const stream = await client.chat.completions.create({
  model: "gpt-4",
  messages: [...],
  stream: true
});

// Collect all content
let fullContent = "";
for await (const chunk of stream) {
  fullContent += chunk.choices[0]?.delta?.content ?? "";
}
console.log(fullContent);

Python

stream = client.chat.completions.create(
    model="gpt-4",
    messages=[...],
    stream=True
)

# Collect all content
full_content = ""
for chunk in stream:
    full_content += chunk.choices[0].delta.content or ""
print(full_content)

Error Handling

TypeScript

try {
  const stream = await client.chat.completions.create({
    model: "gpt-4",
    messages: [...],
    stream: true
  });

  for await (const chunk of stream) {
    process.stdout.write(chunk.choices[0]?.delta?.content ?? "");
  }
} catch (error) {
  if (error instanceof StreamError) {
    console.error("Stream error:", error.message);
  }
}

Python

try:
    stream = client.chat.completions.create(
        model="gpt-4",
        messages=[...],
        stream=True
    )

    for chunk in stream:
        print(chunk.choices[0].delta.content or "", end="")
except StreamError as e:
    print(f"Stream error: {e}")

Note

Streaming responses typically don’t retry automatically. Handle errors by restarting the stream.

Best Practices

1. Use streaming for large responses - AI completions, file processing
2. Handle partial data - Process chunks as they arrive
3. Implement cancellation - Allow users to abort long streams
4. Set appropriate timeouts - Streaming may need longer timeouts
5. Filter sentinel events - Configure sentinel_events to hide them

Next Steps

• Authentication - Configure API authentication
• Pagination - Auto-pagination for list endpoints