trickle

Best-effort parsing of partial / streaming JSON from LLM tool-calls. Zero dependencies, pure stdlib.

The built-in json module is all-or-nothing: feed it a truncated document and it raises. But when an LLM streams a tool-call's arguments, the JSON arrives one chunk at a time — and you usually want to read fields as they appear (to render progressive UI, start fetching, validate early, or fail fast). trickle parses whatever is there: it closes open containers, drops trailing partial tokens, and hands you a normal Python object.

import trickle

trickle.loads('{"city": "Paris", "temp": 1')      # -> {'city': 'Paris', 'temp': 1}
trickle.loads('{"items": ["a", "b')               # -> {'items': ['a', 'b']}
trickle.loads('{"ok": tru')                        # -> {'ok': True}

A complete, valid document parses identically to json.loads. Incomplete input degrades gracefully.

Why

If you've ever written this, you've felt the pain:

try:
    args = json.loads(buffer)   # raises on every chunk until the very last one
except json.JSONDecodeError:
    args = None                 # ...so you see nothing until the stream ends

With trickle the buffer is always parseable, so the structured value materializes field-by-field.

Install

pip install trickle-json

Or just vendor the trickle/ folder — it has no dependencies.

Streaming

Feed chunks to a Stream and read the current value after each one:

from trickle import Stream

s = Stream()
for chunk in llm_token_stream():     # whatever yields text deltas
    value = s.feed(chunk)
    # read a nested field the instant it exists, default until then:
    title = s.get("args", "title", default="…")
    render(title)

final = s.value()                    # complete object once the stream ends

Stream.get(*path, default=None) walks keys/indices safely:

s = Stream('{"user": {"name": "Ada')
s.get("user", "name")        # -> 'Ada'
s.get("user", "age", default=0)   # -> 0  (not there yet)

CLI

# Complete a truncated document on stdin
echo '{"a": 1, "b": [2, 3' | trickle
# -> {"a": 1, "b": [2, 3]}

python -m trickle demo        # narrated streaming walk-through
python -m trickle selftest    # offline assertions, no network/keys

Demo

$ python -m trickle demo
Streaming a tool-call, parsed live after every chunk:

  + '{"tool": "create_e'
    parsed: {'tool': 'create_e'}
    title : <pending>

  + 'tle": "Team sync", "atte'
    parsed: {'tool': 'create_event', 'args': {'title': 'Team sync'}}
    title : Team sync

  + 'ndees": ["ada", "gr'
    parsed: {'tool': 'create_event', 'args': {'title': 'Team sync', 'attendees': ['ada', 'gr']}}
    title : Team sync

  + 'se}}'
    parsed: {'tool': 'create_event', 'args': {'title': 'Team sync', 'attendees': ['ada', 'grace'], 'duration_min': 30, 'all_day': False}}

Final, valid JSON:
  {"tool": "create_event", "args": {"title": "Team sync", "attendees": ["ada", "grace"], "duration_min": 30, "all_day": false}}

API

Function	Description
trickle.loads(s)	Parse a possibly-incomplete JSON string → best-effort Python object.
trickle.complete(s)	Return a valid JSON string for incomplete input.
trickle.loads_at(s)	Like loads, plus the number of chars consumed.
trickle.Stream()	Stateful chunk collector: .feed(chunk), .value(), .get(*path), .raw, .reset(), .feed_iter(chunks).

How it handles truncation

Input	Result
open containers {, [	closed automatically
dangling key {"a": 1, "b"	key dropped → {'a': 1}
dangling colon {"a": 1, "b":	dropped → {'a': 1}
trailing comma [1, 2,	ignored → [1, 2]
partial string "hel	kept → 'hel'
partial number 1.5e / lone -	trimmed to 1.5 / dropped
truncated literal tru, fal, nul	resolved → True, False, None
truncated \uXXXX / dangling \	dropped cleanly
split surrogate pair (emoji)	only emitted once both halves arrive

One caveat to know: a number can read a smaller intermediate value mid-stream — 30 looks like 3 until the next digit (or a delimiter) arrives. Treat scalar values as final only once the field is followed by a ,, }, or ].

License

MIT © Roland Nguyen