Lidando com JSON Parcial em Chamadas de Ferramentas Transmitidas

Quando Claude decide chamar uma ferramenta durante a transmissão, os argumentos da ferramenta não chegam como um único objeto JSON limpo.

Eles chegam como uma sequência de eventos input_json_delta, cada um sendo um pequeno fragmento de string que só se torna JSON válido quando todos os fragmentos foram concatenados em ordem.

Analisar muito cedo - ou analisar cada fragmento individualmente - é uma fonte comum de travamentos em código de uso de ferramentas transmitido. Esta página mostra o padrão correto de acumular e depois analisar.

Receita

import json
import anthropic
 
client = anthropic.Anthropic()
 
tools = [{
    "name": "get_weather",
    "description": "Obtém o clima atual de uma cidade.",
    "input_schema": {
        "type": "object",
        "properties": {"city": {"type": "string"}},
        "required": ["city"],
    },
}]
 
json_buffer = ""
 
with client.messages.stream(
    model="claude-sonnet-5",
    max_tokens=512,
    tools=tools,
    messages=[{"role": "user", "content": "Qual é o clima em Lisboa?"}],
) as stream:
    for event in stream:
        if event.type == "content_block_delta" and event.delta.type == "input_json_delta":
            json_buffer += event.delta.partial_json
        elif event.type == "content_block_stop":
            if json_buffer:
                tool_input = json.loads(json_buffer)
                print(tool_input)  # {"city": "Lisbon"}
                json_buffer = ""

Quando usar isso:

• Qualquer solicitação de transmissão que inclua tools, pois os argumentos de uma chamada de ferramenta sempre são transmitidos como fragmentos, nunca como um único evento.
• Loops de agente que precisam saber o momento exato em que a entrada de uma ferramenta está completa e pronta para ser executada.
• Interfaces de usuário que desejam mostrar "Claude está chamando get_weather..." antes que os argumentos sejam totalmente conhecidos.
• Respostas de múltiplas ferramentas, onde vários blocos tool_use são transmitidos em sequência e cada um precisa de seu próprio buffer.

Exemplo Funcional

Um exemplo completo e executável que lida com várias chamadas de ferramentas em uma única resposta, indexado pelo índice do bloco para que o JSON de cada ferramenta seja acumulado independentemente.

import json
import anthropic
 
client = anthropic.Anthropic()
 
tools = [
    {
        "name": "get_weather",
        "description": "Obtém o clima atual de uma cidade.",
        "input_schema": {
            "type": "object",
            "properties": {"city": {"type": "string"}},
            "required": ["city"],
        },
    },
    {
        "name": "convert_currency",
        "description": "Converte um valor de uma moeda para outra.",
        "input_schema": {
            "type": "object",
            "properties": {
                "amount": {"type": "number"},
                "from_currency": {"type": "string"},
                "to_currency": {"type": "string"},
            },
            "required": ["amount", "from_currency", "to_currency"],
        },
    },
]
 
def stream_tool_calls(prompt: str) -> list[dict]:
    completed_calls = []
    # Indexado pelo índice do bloco de conteúdo, pois vários blocos tool_use podem ser transmitidos em uma única resposta.
    buffers: dict[int, dict] = {}
 
    with client.messages.stream(
        model="claude-sonnet-5",
        max_tokens=1024,
        tools=tools,
        messages=[{"role": "user", "content": prompt}],
    ) as stream:
        for event in stream:
            if event.type == "content_block_start" and event.content_block.type == "tool_use":
                buffers[event.index] = {
                    "name": event.content_block.name,
                    "id": event.content_block.id,
                    "json_parts": [],
                }
            elif event.type == "content_block_delta" and event.delta.type == "input_json_delta":
                if event.index in buffers:
                    buffers[event.index]["json_parts"].append(event.delta.partial_json)
            elif event.type == "content_block_stop" and event.index in buffers:
                entry = buffers.pop(event.index)
                raw_json = "".join(entry["json_parts"]) or "{}"
                completed_calls.append({
                    "id": entry["id"],
                    "name": entry["name"],
                    "input": json.loads(raw_json),
                })
 
    return completed_calls
 
calls = stream_tool_calls("Qual é o clima em Tóquio e converta 50 USD para EUR.")
for call in calls:
    print(f"{call['name']}({call['input']})")

O que isso demonstra:

• O buffer é indexado por event.index, não por uma única string global, para que blocos tool_use concorrentes na mesma resposta não corrompam o JSON uns dos outros.
• A leitura do name e id da ferramenta a partir de content_block_start, pois os eventos input_json_delta carregam apenas fragmentos JSON, não metadados de identificação.
• O adiamento de json.loads(...) até content_block_stop, o único ponto em que a string acumulada tem garantia de estar completa e analisável.
• O fallback para "{}" quando uma chamada de ferramenta não tem argumentos (uma ferramenta sem entrada necessária ainda produz um objeto JSON vazio).

Mergulho Profundo

Como Funciona

• O campo delta.partial_json de cada evento content_block_delta é um fragmento de uma string JSON - não um fragmento de um objeto analisado - portanto, os fragmentos devem ser unidos como strings antes que qualquer análise ocorra.
• content_block_start dispara uma vez por bloco tool_use e carrega o name e um id único da ferramenta, ambos necessários posteriormente para construir o tool_result que você envia de volta para Claude.
• content_block_stop é o sinal de que o JSON de um determinado bloco está completo; analisar antes deste evento falhará na maioria dos fragmentos, pois eles são, individualmente, JSON inválido.
• Múltiplos blocos tool_use em uma resposta são distinguidos por index - o mesmo campo usado para blocos de texto e de pensamento - portanto, qualquer manipulador de fluxo de propósito geral que rastreia o estado por bloco deve usar o índice como chave, não apenas pelo tipo de conteúdo.

partial_json em Resumo

Notas Python

# Uma classe acumuladora pequena e reutilizável em vez de dicionários inline - útil quando você tiver
# mais de alguns locais de chamada fazendo esse padrão.
class ToolCallAccumulator:
    def __init__(self) -> None:
        self._parts: dict[int, list[str]] = {}
        self._meta: dict[int, dict] = {}
 
    def start(self, index: int, name: str, tool_id: str) -> None:
        self._parts[index] = []
        self._meta[index] = {"name": name, "id": tool_id}
 
    def add_delta(self, index: int, fragment: str) -> None:
        self._parts.setdefault(index, []).append(fragment)
 
    def finish(self, index: int) -> dict:
        import json
        raw = "".join(self._parts.pop(index, [])) or "{}"
        meta = self._meta.pop(index)
        return {**meta, "input": json.loads(raw)}

Armadilhas

• Chamar json.loads() em cada input_json_delta individual - quase todos os fragmentos são JSON inválido por si só, então isso gera json.JSONDecodeError constantemente. Correção: acumule fragmentos como strings e analise exatamente uma vez, em content_block_stop.
• Usar um único buffer de string global para todas as chamadas de ferramentas em uma resposta - os deltas de um segundo bloco tool_use corrompem silenciosamente o JSON da primeira ferramenta. Correção: use o event.index como chave do buffer, como mostrado acima.
• Ler content_block.name de um evento delta em vez de content_block_start - os deltas não carregam o nome da ferramenta, apenas content_block_start o faz. Correção: capture name e id quando o bloco começar, armazene-os junto com o buffer.
• Assumir que uma chamada de ferramenta sem argumentos ainda transmite pelo menos um delta - uma ferramenta com um esquema de entrada vazio pode produzir zero eventos input_json_delta. Correção: defina o buffer como "{}" antes de analisar se nenhum delta chegou.
• Esquecer de ecoar o id da ferramenta de volta como tool_use_id no tool_result - Claude não consegue corresponder seu resultado à chamada correta sem ele. Correção: passe o content_block.id através do seu acumulador para o resultado que você envia de volta.
• Tentar "renderizar parcialmente" os argumentos da ferramenta para um usuário enquanto eles são transmitidos - ao contrário do texto, mostrar uma string JSON meio formada raramente é útil e muitas vezes confuso. Correção: mostre um indicador genérico "chamando tool_name..." em vez do JSON parcial bruto.

Alternativas

FAQs

Relacionados

• Como os Eventos Enviados pelo Servidor Potencializam as Respostas Transmitidas do Claude - onde input_json_delta se encaixa no modelo geral de eventos.
• Combinando Transmissão com o Loop de Uso de Ferramentas - o que fazer com uma chamada de ferramenta analisada assim que você a tiver.
• Referência de Tipos de Eventos de Transmissão - a referência completa de eventos/campos.
• Definindo Esquemas de Ferramentas com Nome, Descrição e Esquema de Entrada - como o parâmetro tools usado acima é estruturado.
• Lidando com Transmissão de Uso de Ferramentas em Fluxos de Trabalho Python Assíncronos - o mesmo padrão em código asyncio.

Versões da Pilha: Escrito contra a linha de modelos Claude atual em ~junho de 2026 - Claude Fable 5, Claude Opus 4.8, Claude Sonnet 5 (o padrão), e Claude Haiku 4.5 - e o SDK oficial anthropic para Python (última versão 0.x). Nomes de modelos, versões de SDK e preços mudam rapidamente - verifique os detalhes atuais em platform.claude.com/docs antes de confiar neles.