125. Streaming Responses
Consuming Server-Sent Events (SSE) streaming API responses
125. Streaming Responses
Try it
Raw exercise data
schema_version: '1.0'
id: 125-streaming
concept: streaming responses
se_concept: Consuming Server-Sent Events (SSE) streaming API responses
difficulty: medium
worlds:
ram_manager:
scenario: |
🌊 Aryan wants the LLM to stream tokens so the RAM manager feels
responsive. His streaming parser skips valid data lines, never
yields tokens, and doesn't handle the [DONE] sentinel correctly.
💡 **Fun fact:** Server-Sent Events (SSE) is a W3C standard from 2006 that predates WebSockets by several years. Unlike WebSockets (bidirectional), SSE is one-way — server pushes to client. Every major LLM API (OpenAI, Anthropic, Google Gemini) uses SSE for streaming, because it works over plain HTTP and needs no special infrastructure. The `text/event-stream` MIME type tells browsers and clients to keep the connection open and process lines as they arrive. ChatGPT's live token-by-token output is SSE under the hood.
⚠️ **Watch out:** The SSE `data:` prefix has a mandatory space after the colon — the full prefix is `data: ` (6 characters). `startswith('data:')` (without space) matches 5 chars, so `line[5:]` includes the leading space in the JSON string. `json.loads(' {"type": ...}')` may still work in Python, but it's relying on leniency — always match the full 6-char prefix `'data: '` and slice from index 6 to get clean JSON.
🤔 **Think about it:** `collect_stream` joins tokens with `' '.join()`, adding a space between every token. LLM tokens already contain their own surrounding whitespace and punctuation — the token for a period is `'.'`, not `' .'`. What would `' '.join(['Hello', ',', ' world', '!'])` produce? Why does this cause visible corruption in the output text?
live_dashboard:
scenario: |
📡 A live dashboard streams AI analysis results token-by-token.
The SSE parser corrupts multi-byte UTF-8 characters and fails
silently when JSON lines have unexpected fields.
broken_code: |
import json
from unittest.mock import MagicMock
class _FakeRequests:
"""Simulated requests library for Pyodide environment."""
@staticmethod
def post(url, headers=None, json=None, stream=None, timeout=None):
raise RuntimeError("Real network call — replace with mock in tests")
requests = _FakeRequests()
def stream_llm(prompt: str, api_key: str):
"""Generator that yields text tokens from a streaming LLM response."""
# 💡 : hint : -010 xp without stream=True, requests downloads the entire response body before returning — pass stream=True to requests.post() so iter_lines() delivers bytes as they arrive
# BUG 1: Missing stream=True on requests side — buffers entire response
resp = requests.post(
'https://api.anthropic.com/v1/messages',
headers={
'x-api-key': api_key,
'anthropic-version': '2023-06-01',
'content-type': 'application/json',
},
json={
'model': 'claude-3-haiku-20240307',
'max_tokens': 256,
'stream': True,
'messages': [{'role': 'user', 'content': prompt}],
},
timeout=60,
)
resp.raise_for_status()
for raw_line in resp.iter_lines():
if not raw_line:
continue
line = raw_line.decode('utf-8')
# 💡 : hint : -010 xp SSE lines are formatted as 'data: {...}' with a space — the prefix 'data: ' is 6 characters; use startswith('data: ') and slice line[6:] to get the JSON without a leading space
# BUG 2: Checks for 'data:' without the space — would match 'data:{'
# SSE format is 'data: {...}' with a space after the colon
if not line.startswith('data:'):
continue
# BUG 3: Wrong slice — 'data: ' is 6 chars, slicing from 5 skips the space
# but leaves a leading space in the JSON string, causing JSONDecodeError
payload = line[5:]
if payload == '[DONE]':
break
try:
chunk = json.loads(payload)
# 💡 : hint : -010 xp the Anthropic streaming event type for text tokens is 'content_block_delta', not 'message_delta' — check chunk.get('type') == 'content_block_delta'
# BUG 4: Wrong event type — should be 'content_block_delta'
if chunk.get('type') == 'message_delta':
yield chunk['delta']['text']
except json.JSONDecodeError:
pass
def collect_stream(prompt: str, api_key: str) -> str:
"""Collect all streamed tokens into a single string."""
# 💡 : hint : -010 xp LLM tokens already include their own surrounding whitespace and punctuation — use ''.join() (empty string) not ' '.join() to avoid adding extra spaces between tokens
# BUG 5: Joining with spaces instead of empty string — adds spaces between tokens
return ' '.join(stream_llm(prompt, api_key))
solution_code: |
import json
from unittest.mock import MagicMock
class _FakeRequests:
"""Simulated requests library for Pyodide environment."""
@staticmethod
def post(url, headers=None, json=None, stream=None, timeout=None):
raise RuntimeError("Real network call — replace with mock in tests")
requests = _FakeRequests()
def stream_llm(prompt: str, api_key: str):
"""Generator that yields text tokens from a streaming LLM response."""
resp = requests.post(
'https://api.anthropic.com/v1/messages',
headers={
'x-api-key': api_key,
'anthropic-version': '2023-06-01',
'content-type': 'application/json',
},
json={
'model': 'claude-3-haiku-20240307',
'max_tokens': 256,
'stream': True,
'messages': [{'role': 'user', 'content': prompt}],
},
stream=True, # FIX 1: tell requests not to buffer the full response
timeout=60,
)
resp.raise_for_status()
for raw_line in resp.iter_lines():
if not raw_line:
continue
line = raw_line.decode('utf-8')
if not line.startswith('data: '): # FIX 2: SSE format has a space after colon
continue
payload = line[6:] # FIX 3: slice past 'data: ' (6 chars)
if payload.strip() == '[DONE]':
break
try:
chunk = json.loads(payload)
if chunk.get('type') == 'content_block_delta': # FIX 4: correct event type
yield chunk['delta']['text']
except (json.JSONDecodeError, KeyError):
pass
def collect_stream(prompt: str, api_key: str) -> str:
"""Collect all streamed tokens into a single string."""
return ''.join(stream_llm(prompt, api_key)) # FIX 5: no spaces between tokens
setup: |
import json
from unittest.mock import MagicMock
tests: |
import json
from unittest.mock import MagicMock
def _make_sse_lines(tokens, done=True):
"""Build a list of SSE byte lines from a list of text tokens."""
lines = []
for i, token in enumerate(tokens):
chunk = {
'type': 'content_block_delta',
'index': i,
'delta': {'type': 'text_delta', 'text': token},
}
lines.append(b'data: ' + json.dumps(chunk).encode())
lines.append(b'') # blank line between events
if done:
lines.append(b'data: [DONE]')
return lines
def _mock_streaming_resp(tokens):
r = MagicMock()
r.status_code = 200
r.raise_for_status.return_value = None
r.iter_lines.return_value = iter(_make_sse_lines(tokens))
return r
def test_collect_stream_joins_without_spaces():
mock_resp = _mock_streaming_resp(['Hello', ', ', 'world', '!'])
original_post = requests.post
requests.post = MagicMock(return_value=mock_resp)
try:
result = collect_stream('hi', 'key')
finally:
requests.post = original_post
assert result == 'Hello, world!', (
f'Expected "Hello, world!", got "{result}" — join with empty string, not spaces'
)
def test_stream_llm_yields_all_tokens():
mock_resp = _mock_streaming_resp(['foo', 'bar', 'baz'])
original_post = requests.post
requests.post = MagicMock(return_value=mock_resp)
try:
tokens = list(stream_llm('prompt', 'key'))
finally:
requests.post = original_post
assert tokens == ['foo', 'bar', 'baz']
def test_stream_llm_stops_at_done():
lines = _make_sse_lines(['a', 'b'], done=True)
extra = {
'type': 'content_block_delta',
'index': 99,
'delta': {'type': 'text_delta', 'text': 'c'},
}
lines.append(b'data: ' + json.dumps(extra).encode())
r = MagicMock()
r.status_code = 200
r.raise_for_status.return_value = None
r.iter_lines.return_value = iter(lines)
original_post = requests.post
requests.post = MagicMock(return_value=r)
try:
tokens = list(stream_llm('prompt', 'key'))
finally:
requests.post = original_post
assert 'c' not in tokens, 'Should stop at [DONE], not yield tokens after it'
def test_stream_llm_skips_non_data_lines():
lines = [
b'event: message_start',
b'',
b'data: {"type": "content_block_delta", "delta": {"text": "hi"}}',
b'data: [DONE]',
]
r = MagicMock()
r.status_code = 200
r.raise_for_status.return_value = None
r.iter_lines.return_value = iter(lines)
original_post = requests.post
requests.post = MagicMock(return_value=r)
try:
tokens = list(stream_llm('prompt', 'key'))
finally:
requests.post = original_post
assert tokens == ['hi']
def test_stream_llm_uses_stream_true():
mock_resp = _mock_streaming_resp([])
original_post = requests.post
mock_post = MagicMock(return_value=mock_resp)
requests.post = mock_post
try:
list(stream_llm('p', 'k'))
finally:
requests.post = original_post
_, kwargs = mock_post.call_args
assert kwargs.get('stream') is True, 'Must pass stream=True to requests.post'
se_concept_detail: |
## Streaming Responses (Server-Sent Events)
LLM APIs send responses as they generate them using SSE (Server-Sent Events).
### SSE format
event: message_start data: {“type”: “message_start”, …}
data: {“type”: “content_block_delta”, “delta”: {“text”: “Hello”}}
data: {“type”: “content_block_delta”, “delta”: {“text”: “, world”}}
data: [DONE]
- Lines starting with `data: ` contain JSON payloads
- Empty lines separate events
- `data: [DONE]` signals the end of the stream
### Python implementation
```python
resp = requests.post(url, json=payload, stream=True, timeout=60)
for raw_line in resp.iter_lines():
if not raw_line:
continue
line = raw_line.decode('utf-8')
if not line.startswith('data: '):
continue
payload = line[6:] # skip 'data: ' (6 chars)
if payload == '[DONE]':
break
chunk = json.loads(payload)
if chunk.get('type') == 'content_block_delta':
yield chunk['delta']['text']Collecting to string
# Right: no spaces between tokens
text = ''.join(stream_llm(prompt, key))
# Wrong: adds spaces between every token
text = ' '.join(stream_llm(prompt, key))Printing live
for token in stream_llm(prompt, key):
print(token, end='', flush=True)
print()learning_objectives:
- Parse SSE lines with the correct data prefix (data: with space, 6 chars)
- Pass stream=True to requests.post() for lazy response reading
- Use a generator function with yield to emit tokens one at a time
- Stop at the [DONE] sentinel
- Join tokens with empty string, not spaces
key_concepts:
- Server-Sent Events (SSE) — streaming protocol
- requests stream=True — lazy response reading
- iter_lines() — iterate SSE lines
- ‘’.join() vs ’ ’.join() — token concatenation
- Generator / yield — lazy token emission
resources:
- title: Server-Sent Events — MDN url: https://developer.mozilla.org/en-US/docs/Web/API/Server-sent_events type: reference
- title: Anthropic Streaming API url: https://docs.anthropic.com/en/api/messages-streaming type: official
- title: Python Generators — Real Python url: https://realpython.com/introduction-to-python-generators/ type: tutorial
- title: Corey Schafer — Python Generators Tutorial url: https://www.youtube.com/watch?v=bD05uGo_sVI type: video
- title: requests — Streaming Requests url: https://docs.python-requests.org/en/latest/user/advanced/#streaming-requests type: official
tasks:
- test: test_collect_stream_joins_without_spaces label: collect_stream joins tokens without spaces
- test: test_stream_llm_yields_all_tokens label: stream_llm yields every token
- test: test_stream_llm_stops_at_done label: stream_llm stops at [DONE] sentinel
- test: test_stream_llm_skips_non_data_lines label: stream_llm skips non-data SSE lines
- test: test_stream_llm_uses_stream_true label: stream_llm passes stream=True to requests