chunkstream 🌊

Zero-dependency, memory-efficient streaming chunk buffer, delimiter-aware frame accumulator, and Server-Sent Events (SSE) parser for Python.

Designed specifically for streaming LLM tokens (OpenAI, Anthropic, Gemini, Ollama, DeepSeek), HTTP chunked transfer decoders, WebSocket packet framing, and async network pipelines.

⚡ Why chunkstream?

When consuming real-world streaming APIs, network sockets deliver data in arbitrary packet slices (e.g. {"mess, age": "hel, lo"}). Handling fragmented chunks, multibyte UTF-8 characters across boundaries, and Server-Sent Events without external dependencies or memory bloat is tedious.

chunkstream solves this with:

• Zero Dependencies: Pure standard library, compatible with Python 3.10 through 3.14.
• Dual Sync & Async Buffers: ChunkBuffer for synchronous streams and AsyncChunkBuffer for asyncio event-driven loops.
• Spec-Compliant SSE Parser: Parses data:, event:, id:, and [DONE] markers across fragmented chunks in both parse_sse and parse_sse_async.
• Delimiter & Fixed-Length Framers: Split byte streams by arbitrary delimiters (\n, \r\n\r\n, \0, or multipart boundaries).
• Backpressure & Overflow Protection: Configurable max_size prevents runaway memory allocation on slow consumers.
• Sliding Stream Window: Lookbehind and deduplication window for stream inspection.

📦 Installation

pip install chunkstream

🚀 Quickstart

1. Parsing Streaming LLM Server-Sent Events (SSE)

Seamlessly parse fragmented chunks from any LLM or HTTP client (httpx, aiohttp, urllib):

from chunkstream import parse_sse

def mock_llm_stream():
    yield b'data: {"id": "c1", "choices": [{"delta": {"con'
    yield b'tent": "Hello"}}]}\n\n'
    yield b'data: {"id": "c1", "choices": [{"delta": {"content": " world!"}}]}\n\n'
    yield b'data: [DONE]\n\n'

for event in parse_sse(mock_llm_stream()):
    if event.is_done:
        print("\n[Generation Complete]")
        break
    print(f"Token: {event.data}")

2. Asynchronous SSE Stream Parsing

import asyncio
from chunkstream import parse_sse_async

async def main():
    async for event in parse_sse_async(async_byte_generator()):
        if event.is_done:
            break
        print(event.data)

asyncio.run(main())

3. Delimiter Framing (ChunkBuffer)

Accumulate incoming network chunks and read complete lines or frames:

from chunkstream import ChunkBuffer

buffer = ChunkBuffer(max_size=1024 * 1024) # 1MB limit

buffer.feed(b"frame_one\0frame_two\0partial_th")

# Read frame by frame
while frame := buffer.read_until(b"\0"):
    print("Frame:", frame.decode())

# Remaining data stays safely buffered
print("Unconsumed:", buffer.read()) # b"partial_th"

4. Sliding Stream Window

Track rolling history over an ongoing stream:

from chunkstream import StreamWindow

window = StreamWindow(size=3)
for item in ["A", "B", "C", "D"]:
    window.feed(item)

print(window.snapshot()) # ['B', 'C', 'D']

🧪 Running Tests

pip install pytest
pytest

📄 License

MIT © 2026 Hamidreza (mrhamidrezam)