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.
| 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 |
pip install -e .from streamparse import JSONStreamParser
parser = JSONStreamParser()
# Feed chunks as they arrive — objects are emitted when complete
chunks = ['{"id": 1, "val', 'ue": "he', 'llo"}{"id": 2}']
for chunk in chunks:
for obj in parser.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.
from streamparse import MarkdownStreamParser
parser = MarkdownStreamParser()
tokens = [
"Here's an example:\n",
"```python\n",
"x = 1 + 2\n",
"print(x)\n",
"```\n",
"And some more text.\n",
]
for token in tokens:
for event in parser.feed(token):
if event["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 block
for event in parser.flush():
print(f"[unclosed {event['language']}]\n{event['code']}")from streamparse import ToolCallAccumulator
acc = ToolCallAccumulator()
# Feed raw streaming events (dicts or SDK objects)
for event in stream:
calls = acc.feed(event)
for call in calls:
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.
from streamparse import SSEParser
parser = SSEParser()
raw = b"event: content_block_delta\ndata: {\"type\":\"text\"}\n\ndata: more\n\n"
for event_type, data in parser.feed(raw):
print(event_type, data)
# content_block_delta {"type":"text"}
# message morepytest -q..............................................
46 passed in 0.04s