streamjson

Parse partial / streaming JSON — the best-effort object out of an incomplete string (LLM token streams, SSE), and array items or OpenAI deltas emitted as they complete. Stdlib only.

from streamjson import parse_partial

parse_partial('{"a": 1, "b": [1, 2, 3')      # token limit / mid-stream
# {'a': 1, 'b': [1, 2, 3]}

parse_partial('{"name": "Bob", "bio": "hal')  # unterminated trailing string
# {'name': 'Bob'}                              #   dropped, not guessed

No dependencies. No model call. A tiny tolerant parser plus two streaming helpers.

The problem

json.loads() throws the instant the stream is cut off — which is every intermediate frame of an LLM response. You want the object so far, and you want to act on array items the moment each one finishes, not after the whole payload lands.

streamjson gives you the best partial object: completed pairs and elements are kept, an incomplete trailing string / key / value is dropped, and open containers are implicitly closed.

API

from streamjson import parse_partial, JSONStream, stream_items, iter_sse, accumulate_openai

parse_partial(s) -> object | None

Best-effort Python value from a possibly-incomplete JSON string. None if nothing complete can be recovered.

parse_partial('{"a":1,"b":2,"c":')   # {'a': 1, 'b': 2}   — dangling key dropped
parse_partial('{"a":{"b":[1,2')       # {'a': {'b': [1, 2]}} — containers closed
parse_partial('{"a":1,"ok":tr')       # {'a': 1}           — partial literal dropped

JSONStream — "what's the object so far"

s = JSONStream()
s.feed('{"a": 1, "b":')
s.value()                 # {'a': 1}
s.feed(' 2}')
s.value()                 # {'a': 1, 'b': 2}

stream_items(chunks, key=None) — array items as they finish

Yields each completed element of a top-level array (or the array under key) exactly once, never a partial — even if a chunk boundary lands in the middle of an element.

chunks = ['[{"i":0},{', '"i":1},{"i', '":2}]']
list(stream_items(chunks))                       # [{'i': 0}, {'i': 1}, {'i': 2}]

doc = ['{"items":[{"x":1},', '{"x":2}]}']
list(stream_items(doc, key="items"))             # [{'x': 1}, {'x': 2}]

SSE — iter_sse and accumulate_openai

sse = (
    'data: {"choices":[{"delta":{"content":"Hello"}}]}\n\n'
    'data: {"choices":[{"delta":{"content":" world"}}]}\n\n'
    'data: [DONE]'
)
accumulate_openai(sse)    # 'Hello world'   — the reconstructed message
list(iter_sse(sse))       # the two data payloads ([DONE] skipped)

accumulate_openai is robust to partial / garbage payloads — it parses each event best-effort and concatenates every choices[].delta.content it finds.

CLI

truncated_dump.json | streamjson            # repair a cut-off dump → pretty JSON
streamjson reply.json --items               # each completed array item, one per line
streamjson reply.json --items --key items   # ... for the array under "items"
some-llm-call | streamjson --sse            # reconstruct the assistant message
some-llm-call | streamjson --sse --raw      # each SSE data payload, one per line

Reads a file argument or stdin.

Layout

streamjson/
  partial.py   tolerant recursive-descent parser → best partial object
  stream.py    JSONStream + stream_items (array items as they complete)
  sse.py       iter_sse + accumulate_openai (OpenAI-style SSE)
  cli.py       the `streamjson` command

MIT. Stdlib only — no dependencies, no network, no API keys.