stream-parse

Incremental parsers for streaming LLM output. Handle JSON objects, markdown code blocks, tool-call deltas, and SSE events — all without buffering the full response first.

What's inside

Parser	What it handles
JSONStreamParser	Extracts complete JSON objects from a character stream; depth-tracking state machine handles nested structures and strings correctly
MarkdownStreamParser	Emits code_block events when a fenced block closes; flush() recovers unclosed blocks at end-of-stream
ToolCallAccumulator	Reconstructs full tool calls from input_json_delta streaming events; supports multiple concurrent tool calls
SSEParser	Parses Server-Sent Events; emits (event_type, data) tuples; handles multi-line data fields and reconnection comments

Install

pip install -e .

JSON streaming

fromstreamparseimportJSONStreamParserparser=JSONStreamParser()
# Feed chunks as they arrive — objects are emitted when completechunks= ['{"id": 1, "val', 'ue": "he', 'llo"}{"id": 2}']
forchunkinchunks:
forobjinparser.feed(chunk):
print(obj)
# {"id": 1, "value": "hello"}# {"id": 2}

Handles strings (including escaped quotes and backslashes), nested objects and arrays, and multiple objects in a single chunk. Works correctly with JSON that contains } inside a string value.

Markdown streaming

fromstreamparseimportMarkdownStreamParserparser=MarkdownStreamParser()
tokens= [
"Here's an example:\n",
"```python\n",
"x = 1 + 2\n",
"print(x)\n",
"```\n",
"And some more text.\n",
]
fortokenintokens:
foreventinparser.feed(token):
ifevent["type"] =="code_block":
print(f"[{event['language']}]\n{event['code']}")
else:
print(event["text"], end="")
# Here's an example:# [python]# x = 1 + 2# print(x)# And some more text.# At end-of-stream, flush recovers any unclosed blockforeventinparser.flush():
print(f"[unclosed {event['language']}]\n{event['code']}")

Tool-call delta accumulation

fromstreamparseimportToolCallAccumulatoracc=ToolCallAccumulator()
# Feed raw streaming events (dicts or SDK objects)foreventinstream:
calls=acc.feed(event)
forcallincalls:
print(f"Tool: {call['name']}, Input: {call['input']}")

Handles content_block_start, content_block_delta (input_json_delta), and content_block_stop events. Emits a complete tool call dict when the block closes.

SSE parsing

fromstreamparseimportSSEParserparser=SSEParser()
raw=b"event: content_block_delta\ndata: {\"type\":\"text\"}\n\ndata: more\n\n"forevent_type, datainparser.feed(raw):
print(event_type, data)
# content_block_delta {"type":"text"}# message more

Testing

pytest -q

..............................................
46 passed in 0.04s