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.