Handling partial JSON in streamed LLM tool calls

When you stream tool calls from an LLM, the model emits arguments as a sequence of string deltas rather than one complete payload. Handling partial json streamed tool calls correctly requires accumulating those deltas per tool call, detecting when the JSON is complete, and avoiding crashes on intermediate malformed states. This guide walks through a concrete client-side pattern you can drop into production.

Step 1: Request streaming tool calls

Use an OpenAI-compatible client with stream=True and a tools list. The model returns delta.tool_calls chunks instead of a finished message.tool_calls array.

from openai import OpenAI

client = OpenAI(base_url="https://api.openai.com/v1", api_key="sk-...")

stream = client.chat.completions.create(
    model="gpt-4o",
    messages=[{"role": "user", "content": "Get weather for SF and NYC"}],
    tools=[{
        "type": "function",
        "function": {
            "name": "get_weather",
            "parameters": {
                "type": "object",
                "properties": {
                    "city": {"type": "string"},
                    "units": {"type": "string", "enum": ["metric", "imperial"]}
                },
                "required": ["city"]
            }
        }
    }],
    stream=True,
)

The stream stays open until the model finishes or the connection drops. Each chunk may contain zero or more tool_calls entries.

Step 2: Accumulate deltas by tool call ID

The first delta for a tool call carries an id and usually a function.name. Subsequent deltas for the same call often omit the id and send only function.arguments fragments. Buffer them in a dict keyed by ID, and keep an index map for chunks that arrive without an ID.

tool_buffers = {}   # id -> {"name": str, "arguments": str}
index_to_id = {}    # stream index -> id

for chunk in stream:
    delta = chunk.choices[0].delta
    if not delta.tool_calls:
        continue
    for tc in delta.tool_calls:
        # tc.index is always present; tc.id only on first fragment
        if tc.id:
            tool_buffers[tc.id] = {"name": "", "arguments": ""}
            if tc.function and tc.function.name:
                tool_buffers[tc.id]["name"] = tc.function.name
            index_to_id[tc.index] = tc.id
        else:
            tid = index_to_id[tc.index]
            if tc.function and tc.function.name:
                tool_buffers[tid]["name"] = tc.function.name
        tid = tc.id or index_to_id[tc.index]
        if tc.function and tc.function.arguments:
            tool_buffers[tid]["arguments"] += tc.function.arguments

This structure survives interleaved calls and out-of-order argument fragments.

Step 3: Detect JSON completion without a parser

You cannot call json.loads on a fragment like {"city": "SF because it raises. Before parsing, check brace and bracket balance while respecting string escapes. A correct check ignores braces inside quoted strings.

def json_depth(s: str) -> int:
    depth = 0
    in_str = False
    esc = False
    for ch in s:
        if in_str:
            if esc:
                esc = False
            elif ch == '\\':
                esc = True
            elif ch == '"':
                in_str = False
            continue
        if ch == '"':
            in_str = True
        elif ch in '{[':
            depth += 1
        elif ch in '}]':
            depth -= 1
    return depth

def is_complete_json(s: str) -> bool:
    s = s.strip()
    if not s or s[0] not in '{[':
        return False
    return json_depth(s) == 0

json_depth returns 0 only when every opened object or array has closed. It is not a full validator, but it is enough to know when to attempt a real parse.

Step 4: Parse partial json streamed tool calls for live UI

Even before completion, you may want to show a partially filled form or a tree view. A best-effort parse closes open containers and feeds the result to json.loads. Treat the output as untrusted preview only.

import json

def parse_partial(s: str):
    if not s.strip():
        return None
    try:
        return json.loads(s)
    except json.JSONDecodeError:
        pass
    depth = json_depth(s)
    if depth > 0:
        closer = ']' if s.strip()[0] == '[' else '}'
        try:
            return json.loads(s + closer * depth)
        except json.JSONDecodeError:
            return None
    return None

Use it to render progress:

for tid, buf in tool_buffers.items():
    if is_complete_json(buf["arguments"]):
        continue
    preview = parse_partial(buf["arguments"])
    if preview is not None:
        print(f"Live {buf['name']} args: {preview}")

This handles the common case where the model is mid-object. It will mis-handle nested asymmetries (e.g., an open array inside an open object), but the worst outcome is a None preview, not a crash.

Step 5: Validate and execute only on complete calls

Run a schema check with pydantic (or your own validator) the moment is_complete_json returns true. Never execute a tool on a partial fragment.

from pydantic import BaseModel, ValidationError

class WeatherArgs(BaseModel):
    city: str
    units: str = "metric"

def dispatch(tid: str, buf: dict):
    if not is_complete_json(buf["arguments"]):
        return  # still streaming
    try:
        args = WeatherArgs(**json.loads(buf["arguments"]))
    except (json.JSONDecodeError, ValidationError) as e:
        print(f"Invalid tool call {tid}: {e}")
        return
    # safe to call your function
    result = get_weather(args.city, args.units)
    print(f"Dispatched {buf['name']}: {result}")

If the stream ends and is_complete_json is false, discard the buffer. A truncated call is worse than a missing call.

Step 6: Survive provider interruptions

Network resets and provider rate limits can terminate a stream after a partial delta. If you route through n4n.ai’s OpenAI-compatible endpoint, automatic fallback when a provider is rate-limited or degraded keeps the connection alive across backends, but your accumulator must still discard partial json streamed tool calls that terminate early due to a hard client timeout. Wrap the consumption loop in a timeout and treat any tool call without a completed JSON argument as failed.

import signal

def timeout_handler(signum, frame):
    raise TimeoutError("stream stalled")

signal.signal(signal.SIGALRM, timeout_handler)
signal.alarm(30)  # 30s max

try:
    for chunk in stream:
        # accumulation logic from Step 2
        ...
except TimeoutError:
    pass
finally:
    signal.alarm(0)

# post-loop cleanup
for tid, buf in tool_buffers.items():
    if not is_complete_json(buf["arguments"]):
        print(f"Dropping incomplete tool call {tid}")
        del tool_buffers[tid]

Verify success

Build a unit test that replays canned argument fragments through your accumulator and completion check.

def test_partial_json():
    fragments = ['{"city":', ' "SF",', ' "units":', ' "metric"}']
    buf = ""
    for i, frag in enumerate(fragments):
        buf += frag
        if i < len(fragments) - 1:
            assert not is_complete_json(buf), f"false complete at {i}"
    assert is_complete_json(buf)
    assert json.loads(buf)["city"] == "SF"
    assert parse_partial('{"city":') == {"city": None} or parse_partial('{"city":') is not None

test_partial_json()

Run the full streaming script against a live model and confirm:

1. Tool call names appear before arguments finish.
2. parse_partial prints a growing dict as chunks arrive.
3. dispatch fires exactly once per valid completed call.
4. Killing the network mid-stream logs “Dropping incomplete tool call” instead of raising.

That loop is the baseline for any production UI that displays or acts on streamed tool calls. The same buffers work whether you render React state or feed a queue for background jobs—just keep the completeness check strict and the partial parse forgiving.