Streaming - Anannas

The Responses API supports streaming responses, allowing you to receive output as it’s generated in real-time. This is particularly useful for chat interfaces and applications that need to display responses progressively.

Enabling Streaming

To enable streaming, set stream: true in your request:

import requests
import json

response = requests.post(
    "https://api.anannas.ai/api/v1/responses",
    headers={
        "Authorization": "Bearer <ANANNAS_API_KEY>",
        "Content-Type": "application/json",
    },
    json={
        "model": "openai/gpt-5-mini",
        "input": "Tell me a story about a robot.",
        "stream": True,
        "max_output_tokens": 2000,
    },
    stream=True,
)

for line in response.iter_lines():
    if line:
        line_text = line.decode('utf-8')
        if line_text.startswith('data: '):
            data_str = line_text[6:]
            if data_str == '[DONE]':
                break
            try:
                event = json.loads(data_str)
                if event.get('type') == 'response.output_text.delta':
                    delta = event.get('data', {}).get('delta', '')
                    if delta:
                        print(delta, end='', flush=True)
            except json.JSONDecodeError:
                pass

const response = await fetch('https://api.anannas.ai/api/v1/responses', {
  method: 'POST',
  headers: {
    'Authorization': 'Bearer <ANANNAS_API_KEY>',
    'Content-Type': 'application/json',
  },
  body: JSON.stringify({
    model: 'openai/gpt-5-mini',
    input: 'Tell me a story about a robot.',
    stream: true,
    max_output_tokens: 2000,
  }),
});

const reader = response.body?.getReader();
const decoder = new TextDecoder();

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

    const chunk = decoder.decode(value);
    const lines = chunk.split('\n');

    for (const line of lines) {
      if (line.startsWith('data: ')) {
        const dataStr = line.slice(6);
        if (dataStr === '[DONE]') {
          break;
        }
        try {
          const event = JSON.parse(dataStr);
          if (event.type === 'response.output_text.delta') {
            const delta = event.data?.delta || '';
            if (delta) {
              process.stdout.write(delta);
            }
          }
        } catch (e) {
          // Skip invalid JSON
        }
      }
    }
  }
}

Stream Event Types

The streaming API uses Server-Sent Events (SSE) format. Each event has a type field indicating the event type: Event Types:

response.created: Initial response creation event
response.output_item.added: New output item added
response.output_text.delta: Text delta (incremental text)
response.output_item.done: Output item completed
response.completed: Response generation completed

Example Stream Events:

data: {"type":"response.created","data":{"id":"resp_123","created_at":1693350000,"model":"openai/gpt-5-mini"}}

data: {"type":"response.output_item.added","data":{"output_index":0,"item":{"type":"message","role":"assistant"}}}

data: {"type":"response.output_text.delta","data":{"output_index":0,"delta":"Once"}}

data: {"type":"response.output_text.delta","data":{"output_index":0,"delta":" upon"}}

data: {"type":"response.output_text.delta","data":{"output_index":0,"delta":" a"}}

data: {"type":"response.output_text.delta","data":{"output_index":0,"delta":" time"}}

data: {"type":"response.output_item.done","data":{"output_index":0}}

data: {"type":"response.completed","data":{"id":"resp_123"}}

data: [DONE]

Processing Stream Events

Here’s a more complete example that handles all event types:

import requests
import json

response = requests.post(
    "https://api.anannas.ai/api/v1/responses",
    headers={
        "Authorization": "Bearer <ANANNAS_API_KEY>",
        "Content-Type": "application/json",
    },
    json={
        "model": "openai/gpt-5-mini",
        "input": "Explain quantum computing.",
        "stream": True,
        "max_output_tokens": 2000,
    },
    stream=True,
)

full_text = ""

for line in response.iter_lines():
    if line:
        line_text = line.decode('utf-8')
        
        # Skip heartbeat messages
        if line_text.startswith(':ANANNAS PROCESSING'):
            continue
            
        if line_text.startswith('data: '):
            data_str = line_text[6:]
            
            if data_str == '[DONE]':
                print("\n\nStream completed.")
                break
                
            try:
                event = json.loads(data_str)
                event_type = event.get('type')
                event_data = event.get('data', {})
                
                if event_type == 'response.created':
                    print(f"Response ID: {event_data.get('id')}")
                    print(f"Model: {event_data.get('model')}\n")
                    
                elif event_type == 'response.output_text.delta':
                    delta = event_data.get('delta', '')
                    if delta:
                        full_text += delta
                        print(delta, end='', flush=True)
                        
                elif event_type == 'response.output_item.done':
                    print("\n\nOutput item completed.")
                    
                elif event_type == 'response.completed':
                    print("\n\nResponse generation completed.")
                    
            except json.JSONDecodeError:
                pass

print(f"\n\nFull text: {full_text}")

const response = await fetch('https://api.anannas.ai/api/v1/responses', {
  method: 'POST',
  headers: {
    'Authorization': 'Bearer <ANANNAS_API_KEY>',
    'Content-Type': 'application/json',
  },
  body: JSON.stringify({
    model: 'openai/gpt-5-mini',
    input: 'Explain quantum computing.',
    stream: true,
    max_output_tokens: 2000,
  }),
});

const reader = response.body?.getReader();
const decoder = new TextDecoder();
let fullText = '';

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

    const chunk = decoder.decode(value);
    const lines = chunk.split('\n');

    for (const line of lines) {
      // Skip heartbeat messages
      if (line.startsWith(':ANANNAS PROCESSING')) {
        continue;
      }

      if (line.startsWith('data: ')) {
        const dataStr = line.slice(6);

        if (dataStr === '[DONE]') {
          console.log('\n\nStream completed.');
          break;
        }

        try {
          const event = JSON.parse(dataStr);
          const eventType = event.type;
          const eventData = event.data || {};

          if (eventType === 'response.created') {
            console.log(`Response ID: ${eventData.id}`);
            console.log(`Model: ${eventData.model}\n`);
          } else if (eventType === 'response.output_text.delta') {
            const delta = eventData.delta || '';
            if (delta) {
              fullText += delta;
              process.stdout.write(delta);
            }
          } else if (eventType === 'response.output_item.done') {
            console.log('\n\nOutput item completed.');
          } else if (eventType === 'response.completed') {
            console.log('\n\nResponse generation completed.');
          }
        } catch (e) {
          // Skip invalid JSON
        }
      }
    }
  }
}

console.log(`\n\nFull text: ${fullText}`);

Heartbeat Messages

The streaming API sends heartbeat messages (:ANANNAS PROCESSING) to keep the connection alive. These should be ignored when processing events.

Error Handling in Streams

Errors in streaming responses are sent as events:

{
  "error": {
    "message": "Rate limit exceeded",
    "type": "rate_limit_error",
    "code": "rate_limit_exceeded"
  }
}

Always check for error events in your stream processing:

for line in response.iter_lines():
    if line:
        line_text = line.decode('utf-8')
        if line_text.startswith('data: '):
            data_str = line_text[6:]
            try:
                event = json.loads(data_str)
                
                # Check for errors
                if 'error' in event:
                    print(f"Error: {event['error']['message']}")
                    break
                    
                # Process normal events
                # ...
            except json.JSONDecodeError:
                pass

try {
  const event = JSON.parse(dataStr);
  
  // Check for errors
  if (event.error) {
    console.error(`Error: ${event.error.message}`);
    break;
  }
  
  // Process normal events
  // ...
} catch (e) {
  // Handle parse errors
}

Best Practices

Handle heartbeats: Ignore :ANANNAS PROCESSING messages
Check for [DONE]: Stop processing when you receive [DONE]
Error handling: Always check for error events in the stream
Buffer management: For long streams, consider buffering and processing in chunks
Connection management: Handle connection drops gracefully and implement retry logic

Was this page helpful?

​Enabling Streaming

​Stream Event Types

​Processing Stream Events

​Heartbeat Messages

​Error Handling in Streams

​Best Practices