Claude Tool Streaming: Assemble JSON Before Dispatch

Claude Fine-Grained Tool Streaming: Finish the JSON Before Dispatch

The first fragment arrives quickly. A progress indicator moves, the tool name is visible, and the application starts preparing its work. Then the dispatcher receives an empty object. Later fragments contain the document it needed, but the job has already failed. This is an illustrative incident, not a report of an observed provider failure. It captures a specific integration mistake: treating the beginning of a streamed Claude tool block as an executable request.

Fine-grained tool streaming makes input available earlier, but earlier visibility does not mean earlier permission to execute. The application needs two separate milestones: data has started arriving, and a complete tool request has passed validation. This article follows those milestones through a native Claude Messages stream, with an accumulator that holds execution until the message is complete. The example uses a local, read-only handler so the dispatch boundary can be inspected without sending a message, changing a record, or writing a remote file.

Incident note: the empty input was a placeholder

In the hypothetical trace, the dispatcher subscribed to content_block_start. It saw a tool name, a tool identifier, and input: {}, then treated the block as a ready-to-run object. That empty input was not the completed payload. Anthropic’s fine-grained streaming guide explains that actual tool input arrives through JSON string fragments. They must be accumulated and parsed when the block closes; fine-grained mode does not provide server-side JSON validation.

This failure has a recognizable diagnostic signature. A tool attempt exists before the final input fragment. The handler log shows missing required fields. The network trace later contains those fields, so the model appears to have supplied them after the error. Increasing the response timeout does not repair that ordering. Nor does changing the tool description. The investigation should begin at the exact event that caused dispatch and the exact object passed to the handler.

Do not confuse this with a tool that received a complete but unacceptable argument. A finished JSON object containing an unknown release identifier is a validation or authorization issue. An empty start placeholder is an assembly issue. The existing strict tool-calling guide discusses the distinction between shape and business validity in another API family. Claude’s streaming case adds a prior question: has the argument object actually finished arriving?

The timeline that the dispatcher must respect

The streaming Messages documentation defines the message and content-block event structure. The table adds an application policy: wait for the whole message before dispatch. This is deliberately conservative. It prevents a valid-looking early block from triggering work when the rest of the response fails. A system with a different latency requirement can choose another policy, but it should document the consequences rather than accidentally inheriting them from an event callback.

Maintain one buffer per content-block index. The index describes placement within a message; the tool identifier connects the eventual result to the requested call. They have different jobs. Do not use a single global string for every delta, and do not attach text deltas or thinking deltas to the JSON buffer. A message can contain several kinds of content. The dispatcher should become eligible only because its own tool block reached the required state, not because some unrelated content finished.

Transport chunks are another boundary. A TCP read is not an SSE event, and an SSE event is not a finished JSON argument. Use an SDK or a proper SSE parser to obtain typed events before applying this accumulator. Splitting incoming bytes on an arbitrary newline and passing them directly to the tool handler combines three different protocols. That approach often appears to work on a short demo and then fails when escape sequences or longer parameter values span boundaries.

Enable the current field on the intended tool

Current Anthropic documentation exposes eager_input_streaming: true on a user-defined tool, together with streaming on the request. The tool reference identifies the field and its scope. The feature guide says all Claude API models support fine-grained streaming. The per-tool field replaces reliance on the older beta header; this example uses the field directly and does not add a legacy header.

Enable it selectively. A tool whose input is a large document can provide meaningful receiving progress before the document is complete. A tool with a two-character country code may offer little practical benefit. Measure what the application actually improves: time to the first useful progress update, time to a validated input, and time to the handler result. The first-visible-output timing article explains why the first network event alone is an incomplete user-experience metric.

At a gateway boundary, ask a narrow compatibility question: does this route retain the per-tool field and deliver native input JSON deltas with their indices and identifiers? A successful ordinary text response cannot answer that. An OpenAI-compatible response shape also does not establish native Messages event compatibility. Keep the model identifier, endpoint, authentication method, SDK version, and observed event types with the test record. Those facts make a failure report actionable without assuming which layer caused it.

A native reference accumulator with a read-only dispatcher

This standalone script uses the Anthropic SDK’s raw event stream rather than a helper that presents already assembled tool input. It requests a local inspection of release notes, buffers every argument fragment, and calls a handler only after the message has ended correctly. Set RELEASE_NOTES to a local UTF-8 text file and CLAUDE_MODEL to a model available to your native account. The Messages API reference is the request contract.

# Native reference example. Requires the anthropic package.
import json
import os
from pathlib import Path
from anthropic import Anthropic

LIMIT = 2_000_000


def unique_object(pairs):
    result = {}
    for key, value in pairs:
        if key in result:
            raise ValueError("Duplicate JSON key")
        result[key] = value
    return result


def reject_constant(value):
    raise ValueError(f"Non-JSON constant: {value}")


def validate(value):
    if not isinstance(value, dict) or set(value) != {"release_id", "notes"}:
        raise ValueError("Unexpected arguments")
    if value["release_id"] != "fixture-release":
        raise ValueError("Unknown release")
    if not isinstance(value["notes"], str) or not value["notes"]:
        raise ValueError("notes must be nonempty text")
    return value


def inspect_release(value):
    # Read-only demonstration; no external system is contacted.
    return {"release_id": value["release_id"],
            "line_count": len(value["notes"].splitlines())}


notes = Path(os.environ["RELEASE_NOTES"]).read_text(encoding="utf-8")
if len(notes.encode("utf-8")) > LIMIT // 2:
    raise ValueError("Use a smaller first fixture")
client = Anthropic(
    api_key=os.environ["ANTHROPIC_API_KEY"],
    base_url="https://api.anthropic.com",
    timeout=120.0, max_retries=0)
blocks = {}
stop_reason = None
message_complete = False
stream = client.messages.create(
    model=os.environ["CLAUDE_MODEL"], max_tokens=6000, stream=True,
    tools=[{"name": "inspect_release",
            "description": "Inspect supplied release notes without changing anything.",
            "eager_input_streaming": True,
            "input_schema": {"type": "object", "properties": {
                "release_id": {"type": "string"},
                "notes": {"type": "string"}},
                "required": ["release_id", "notes"],
                "additionalProperties": False}}],
    messages=[{"role": "user", "content":
        "Call inspect_release for release_id fixture-release. "
        "Pass these release notes as data in notes, preserving their text:\n" + notes}])
with stream:
    for event in stream:
        if event.type == "error":
            raise RuntimeError("Upstream stream error; dispatch prohibited")
        if event.type == "content_block_start":
            block = event.content_block
            if block.type == "tool_use":
                if event.index in blocks:
                    raise ValueError("Repeated block index")
                blocks[event.index] = {"id": block.id, "name": block.name,
                    "parts": [], "bytes": 0, "closed": False, "value": None}
        elif event.type == "content_block_delta":
            if event.delta.type == "input_json_delta":
                item = blocks[event.index]
                if item["closed"]:
                    raise ValueError("Delta after block closure")
                fragment = event.delta.partial_json
                item["bytes"] += len(fragment.encode("utf-8"))
                if item["bytes"] > LIMIT:
                    raise ValueError("Tool input exceeds local bound")
                item["parts"].append(fragment)
        elif event.type == "content_block_stop" and event.index in blocks:
            item = blocks[event.index]
            if item["closed"]:
                raise ValueError("Repeated block closure")
            raw = "".join(item["parts"])
            value = json.loads(raw, object_pairs_hook=unique_object,
                               parse_constant=reject_constant)
            item["value"] = validate(value)
            item["closed"] = True
        elif event.type == "message_delta":
            reason = event.delta.stop_reason
            if reason is not None:
                stop_reason = reason
        elif event.type == "message_stop":
            message_complete = True

if not message_complete or stop_reason != "tool_use" or not blocks:
    raise RuntimeError("No completed tool-use message; nothing dispatched")
if not all(b["closed"] for b in blocks.values()):
    raise RuntimeError("An input block remains open")
ids = [b["id"] for b in blocks.values()]
if len(ids) != len(set(ids)):
    raise ValueError("Duplicate tool identifier")
if any(b["name"] != "inspect_release" for b in blocks.values()):
    raise ValueError("Unknown tool name")
for index in sorted(blocks):
    block = blocks[index]
    result = inspect_release(block["value"])
    print(json.dumps({"tool_use_id": block["id"], "result": result}))

The script intentionally fails closed on malformed JSON. The Python JSON documentation describes decoder behavior, including customization hooks. Here, duplicate keys and nonstandard constants are rejected so that a downstream parser cannot reinterpret the same input differently. The byte bound limits this example’s buffering, while schema and local value checks protect the handler contract. A production parser should apply limits appropriate to its workload rather than copying the example’s number as a universal allowance.

The prompt asks the model to call the tool, but the code does not assume it will. A normal text answer is a valid model behavior and an unsuccessful tool-streaming fixture. The application reports that outcome instead of manufacturing a tool call. Likewise, the read-only handler only counts lines; it does not verify release quality. Its purpose is to make it obvious when a complete argument object crossed into application execution.

What the failure branches should preserve

If JSON parsing fails, retain diagnostic metadata and an access-controlled copy of the raw fragments when your data policy permits it. Do not repair a missing closing quote or bracket automatically and run the resulting object. The last visible characters may belong to a string value, and a syntactic repair can silently change the requested operation. The upstream guide describes returning an error tool result for invalid input. Implement that as a separate recovery path with a valid conversation record, not as a shortcut around local validation.

If the connection disappears before message completion, no handler has run in this design. A retry can therefore start a fresh model request without duplicating this example’s local action. Production tools still need their own execution records because a failure after dispatch is a different situation. The distinction is easier to maintain when your session event record stores assembly, validation, authorization, dispatch, and result as separate events rather than one vague “tool success” flag.

When the same user action is retried, assign it a stable application operation identifier. A model-generated tool identifier belongs to one response; a fresh response may contain a different identifier for the same requested work. That difference matters once the handler performs side effects. For this article’s read-only inspection it is mostly an observability issue, but the execution boundary should be designed before substituting a handler that changes an external system.

Track elapsed times at assembly start, block closure, message completion, and handler completion. These timestamps reveal whether eager input improves perceived progress while leaving executable-input latency unchanged. They also expose buffering elsewhere in the path: if the client receives all fragments together, the upstream feature may be active while an intermediary still batches delivery. Treat that observation as a transport finding, then inspect the route. Do not infer that the model ignored the field merely because the interface updated once.

A connection retry must create a new accumulator. Appending a replacement response to the previous response’s partial buffer can produce a plausible but incorrect document, especially when both begin with the same keys. Keep the first attempt’s diagnostics under its own request identifier, discard its execution eligibility, and start the replacement at an empty state. If a frontend reconnects to your application’s progress channel, distinguish replaying your stored progress events from restarting the upstream model request. Those are different operations, even when the same spinner remains on screen.

Replay the awkward fragments before a live trial

A useful local fixture deliberately splits JSON inside a string, immediately after a backslash, and between items in an array. None of those individual fragments should be parseable on its own, and none should trigger the handler. Add a second tool block with a different index, an unexpected field, a duplicate key, and a missing final event. The expected outcome is a state transition or rejection, not a fabricated model benchmark. Local replay tests can validate these mechanics without making paid API calls.

Also test the interface that displays progress. It may show bytes received or a generic “receiving tool input” state, but it should not display partial argument text as a completed command. Sensitive notes should not leak into general application logs merely because they arrived incrementally. The audit-trail discussion is useful for deciding which operational metadata belongs in ordinary logs and which content needs restricted access.

Protocol-specific handling matters. The existing Responses tool-call migration tutorial covers a different event contract, while Gemini continuation state introduces another kind of opaque protocol data. Those examples should not be merged into a universal callback that assumes every model emits executable arguments at block start. Keep this Claude accumulator’s assumptions visible and test them at the actual transport boundary.

For a 4SAPI trial, use the integration overview to orient the account setup, then inspect the current API documentation and the routes available from your account. Confirm native Messages streaming and the per-tool field before adapting this example with a separately scoped gateway credential. Start with one harmless release-note fixture. Success means the complete, validated object reaches the handler once, after the required completion events.

Primary references

• Anthropic: Fine-grained tool streaming
• Anthropic: Tool reference
• Anthropic: Streaming messages
• Anthropic: Create a Message
• Python: JSON encoder and decoder