Lesson 4: Streaming Structured Data

Why streaming JSON is usually unsafe

Streaming text works because a fragment of text is useful. A fragment of JSON is not.

{"title": "Spaghetti Carb

That is not valid JSON, and it is indistinguishable from a truncated response. Worse, a partial value can be misleading in a way partial text is not: a stream that has emitted {"servings": 1 may be about to emit 2, and a consumer that parsed and used 1 has acted on a number that was never the answer.

The core problem is that JSON has no meaningful prefix. Validity is a property of the whole document. A structured response should be validated as a unit, which is what Module 6 built.

The default position: do not act on partial JSON. Buffer the stream, validate at the end, and use streaming for the operational benefits rather than for progressive consumption.

Streaming still earns its place even when you buffer, for three reasons. You detect a stall early rather than at the total timeout. You can cancel mid-generation and stop paying, which Lesson 5 covers. And you get usage information from the final chunk.

Partial-JSON parsing with a strict final pass

When progressive display genuinely matters, such as showing extracted fields as they appear, partial parsing is possible with strict conditions.

async def stream_extraction(
    events: AsyncIterator[StreamEvent],
    schema: type[BaseModel],
    on_partial: Callable[[dict], Awaitable[None]],
) -> BaseModel:
    """Emit best-effort partial views, but return only a fully validated model."""
    buffer: list[str] = []

    async for event in events:
        if event.kind != "content":
            continue
        buffer.append(event.text)

        partial = try_parse_partial("".join(buffer))
        if partial is not None:
            await on_partial(partial)          # display only, never act on it

    raw = "".join(buffer)
    return schema.model_validate_json(extract_json_text(raw))     # strict, final

Three rules make this safe.

Partial output is for display only. Never store it, never pass it to another system, and never make a decision on it. A field visible in a partial parse can still change before the document closes.

The final validation is strict and non-negotiable. The last line is the same validation from Module 6, including the repair loop when it fails. Partial parsing is a display convenience layered on top, not a replacement.

Fields that appear late are not present early. Progressive display naturally shows fields in generation order, so a user watching may see an incomplete picture. Design the interface so that partial state is visibly partial.

There are libraries for tolerant JSON parsing that complete open structures. Use one rather than writing your own, and treat its output as a guess, exactly as Module 6 advised.

Reassembling streamed tool-call arguments

This is the case where getting it wrong causes real damage rather than a display glitch.

When a model requests a tool during a stream, the arguments arrive as fragments across many events:

{"type":"tool_use_start","id":"call_01","name":"search_recipes"}
{"type":"tool_args_delta","id":"call_01","partial":"{\"que"}
{"type":"tool_args_delta","id":"call_01","partial":"ry\": \"carbo"}
{"type":"tool_args_delta","id":"call_01","partial":"nara\", \"max_results\": 5}"}
{"type":"tool_use_stop","id":"call_01"}

A tool must never be dispatched before its arguments are complete. Executing on {"query": "carbo is executing on a value that was never requested, and if the tool writes, deletes, or sends anything, the consequence is real.

@dataclass
class PendingToolCall:
    id: str
    name: str
    fragments: list[str] = field(default_factory=list)
    complete: bool = False

    def arguments(self) -> dict:
        return json.loads("".join(self.fragments))


class ToolCallAccumulator:
    """Collects streamed tool call fragments, releasing only completed calls."""

    def __init__(self) -> None:
        self._pending: dict[str, PendingToolCall] = {}

    def handle(self, event: dict) -> PendingToolCall | None:
        """Return a call only once it is complete."""
        match event:
            case {"type": "tool_use_start", "id": str(call_id), "name": str(name)}:
                self._pending[call_id] = PendingToolCall(id=call_id, name=name)
                return None

            case {"type": "tool_args_delta", "id": str(call_id), "partial": str(part)}:
                pending = self._pending.get(call_id)
                if pending is not None:
                    pending.fragments.append(part)
                return None

            case {"type": "tool_use_stop", "id": str(call_id)}:
                pending = self._pending.pop(call_id, None)
                if pending is None:
                    return None
                pending.complete = True
                return pending

        return None

    def incomplete(self) -> list[PendingToolCall]:
        """Calls still pending when the stream ended. These must not run."""
        return list(self._pending.values())

Four details.

Fragments are keyed by call id, because parallel tool calls interleave in the stream. Appending to a single buffer would splice two calls' arguments together.

Nothing is returned until the stop event arrives. The accumulator physically cannot hand you an incomplete call.

A stream ending with pending calls is an error. incomplete() exists so the caller can detect it and treat it as a truncated response rather than silently dropping the call.

The completed arguments still go through Module 6's validation before execution. Complete is not the same as valid, and the model remains an untrusted caller.

[IMAGE PROMPT M8-5
Purpose: Show why streamed tool call arguments must be accumulated by call id and released only when complete.
Visual type: Stream timeline feeding an accumulator with a gated output.
Prompt: A clean educational diagram. Along the top, a horizontal stream of eight event blocks reading left to right, labelled in order: "start id=A", "args A: {"que", "start id=B", "args B: {"pa", "args A: ry":"x"}", "stop id=A", "args B: th":"y"}", "stop id=B". Arrows lead down from these into a container labelled "accumulator", drawn with two separate labelled buffers inside reading "call A buffer" and "call B buffer", each showing its fragments collected in order. From the bottom of the container, two arrows pass through a labelled gate reading "released only on stop event" to two output boxes labelled "dispatch A with complete arguments" and "dispatch B with complete arguments". To the left of the gate, a blocked arrow with a cross marker is labelled "never dispatch a partial buffer".
Required elements: Eight interleaved stream events showing two calls mixed, two separate keyed buffers inside the accumulator, a release gate, two complete dispatch outputs, a blocked partial-dispatch arrow.
Style: Clean educational illustration, professional, uncluttered, high contrast, flat vector.
Layout: Horizontal stream across the top, downward flow into the accumulator, gated outputs at the bottom.
Text labels: "start id=A", "args A", "start id=B", "args B", "stop id=A", "stop id=B", "accumulator", "call A buffer", "call B buffer", "released only on stop event", "dispatch A with complete arguments", "dispatch B with complete arguments", "never dispatch a partial buffer".
Aspect ratio: 16:9
Accessibility: Distinguish the two calls by their id labels rather than colour, and mark the blocked path with a cross plus text.
Avoid: Code screenshots, decorative icons, tiny text, logos, watermarks.
Alt text: Diagram showing two interleaved tool calls arriving as fragments in one stream, collected into separate buffers keyed by call id, and released for dispatch only when each call's stop event arrives.
END IMAGE PROMPT]