> ## Documentation Index
> Fetch the complete documentation index at: https://docs.anannas.ai/llms.txt
> Use this file to discover all available pages before exploring further.
# Streaming
> Receive responses in real-time with streaming.
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:
```python Python theme={null}
import requests
import json
response = requests.post(
"https://api.anannas.ai/api/v1/responses",
headers={
"Authorization": "Bearer ",
"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
```
```typescript TypeScript theme={null}
const response = await fetch('https://api.anannas.ai/api/v1/responses', {
method: 'POST',
headers: {
'Authorization': 'Bearer ',
'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:
```python Python theme={null}
import requests
import json
response = requests.post(
"https://api.anannas.ai/api/v1/responses",
headers={
"Authorization": "Bearer ",
"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}")
```
```typescript TypeScript theme={null}
const response = await fetch('https://api.anannas.ai/api/v1/responses', {
method: 'POST',
headers: {
'Authorization': 'Bearer ',
'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:
```json theme={null}
{
"error": {
"message": "Rate limit exceeded",
"type": "rate_limit_error",
"code": "rate_limit_exceeded"
}
}
```
Always check for error events in your stream processing:
```python Python theme={null}
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
```
```typescript TypeScript theme={null}
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
1. **Handle heartbeats**: Ignore `:ANANNAS PROCESSING` messages
2. **Check for \[DONE]**: Stop processing when you receive `[DONE]`
3. **Error handling**: Always check for error events in the stream
4. **Buffer management**: For long streams, consider buffering and processing in chunks
5. **Connection management**: Handle connection drops gracefully and implement retry logic
Was this page helpful?