Manejo de JSON Parcial Durante Llamadas a Herramientas Transmitidas

Cuando Claude decide llamar a una herramienta mientras transmite, los argumentos de la herramienta no llegan como un objeto JSON limpio.

Llegan como una secuencia de eventos input_json_delta, cada uno un pequeño fragmento de cadena que solo se convierte en JSON válido una vez que cada fragmento ha sido concatenado en orden.

Analizar demasiado pronto, o analizar cada fragmento individualmente, es una fuente común de fallos en el código de uso de herramientas de transmisión. Esta página muestra el patrón correcto de acumular y luego analizar.

Receta

import json
import anthropic
 
client = anthropic.Anthropic()
 
tools = [{
    "name": "get_weather",
    "description": "Get the current weather for a city.",
    "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": "What's the weather in Lisbon?"}],
) 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 = ""

Cuándo usarlo:

Cualquier solicitud de transmisión que incluya tools, ya que los argumentos de una llamada a herramienta siempre se transmiten como fragmentos, nunca como un solo evento.
Bucles agénticos que necesitan saber el momento exacto en que la entrada de una herramienta está completa y lista para ejecutarse.
Interfaces de usuario que desean mostrar "Claude está llamando a get_weather..." antes de que los argumentos se conozcan por completo.
Respuestas de múltiples herramientas, donde varios bloques tool_use se transmiten en secuencia y cada uno necesita su propio búfer.

Ejemplo de trabajo

Un ejemplo completo y ejecutable que maneja múltiples llamadas a herramientas en una respuesta, indexadas por el índice de bloque para que el JSON de cada herramienta se acumule de forma independiente.

import json
import anthropic
 
client = anthropic.Anthropic()
 
tools = [
    {
        "name": "get_weather",
        "description": "Get the current weather for a city.",
        "input_schema": {
            "type": "object",
            "properties": {"city": {"type": "string"}},
            "required": ["city"],
        },
    },
    {
        "name": "convert_currency",
        "description": "Convert an amount from one currency to another.",
        "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 por el índice del bloque de contenido, ya que múltiples bloques tool_use pueden transmitirse en una respuesta.
    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("What's the weather in Tokyo, and convert 50 USD to EUR.")
for call in calls:
    print(f"{call['name']}({call['input']})")

Lo que esto demuestra:

Almacenamiento en búfer indexado por event.index, no una única cadena global, para que los bloques tool_use concurrentes en la misma respuesta no corrompan el JSON del otro.
Lectura del name y id de la herramienta desde content_block_start, ya que los eventos input_json_delta solo llevan fragmentos JSON, no metadatos de identificación.
Aplazamiento de json.loads(...) hasta content_block_stop, el único punto en el que la cadena acumulada está garantizada de ser completa y analizable.
Volver a "{}" cuando una llamada a herramienta no tiene argumentos (una herramienta sin entrada requerida aún produce un objeto JSON vacío).

Análisis Profundo

Cómo funciona

El campo delta.partial_json de cada evento content_block_delta es un fragmento de una cadena JSON, no un fragmento de un objeto analizado, por lo que los fragmentos deben unirse como cadenas antes de que se realice cualquier análisis.
content_block_start se activa una vez por bloque tool_use y lleva el name de la herramienta y un id único, ambos necesarios más tarde para construir el tool_result que se envía de vuelta a Claude.
content_block_stop es la señal de que el JSON de un bloque dado está completo; el análisis antes de este evento fallará en la mayoría de los fragmentos, ya que son, individualmente, JSON inválido.
Múltiples bloques tool_use en una respuesta se distinguen por index, el mismo campo utilizado para los bloques de texto y pensamiento, por lo que cualquier manejador de flujo de propósito general que rastree el estado por bloque debe indexar por índice, no solo por tipo de contenido.

`partial_json` de un vistazo

Campo	Tipo	Significado
`event.delta.partial_json`	`str`	El siguiente fragmento de la cadena JSON de entrada de la herramienta: añadir, nunca reemplazar.
`event.content_block.name`	`str`	La herramienta que se está llamando, disponible en `content_block_start`, no en cada delta.
`event.content_block.id`	`str`	El `tool_use_id` que debe repetirse en el bloque `tool_result` correspondiente.
`event.index`	`int`	A qué bloque de contenido pertenece este evento: la clave de acumulación.

Notas de Python

# Una pequeña clase acumuladora reutilizable en lugar de diccionarios en línea, útil una vez que se tienen
# más de un par de sitios de llamada que usan este patrón.
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)}

Errores comunes

Llamar a json.loads() en cada input_json_delta individual - casi todos los fragmentos son JSON inválido por sí solos, por lo que esto genera json.JSONDecodeError constantemente. Solución: acumular fragmentos como cadenas y analizar exactamente una vez, en content_block_stop.
Usar un búfer de cadena global para todas las llamadas a herramientas en una respuesta - los deltas de un segundo bloque tool_use corrompen silenciosamente el JSON de la primera herramienta. Solución: indexar el búfer por event.index, como se muestra arriba.
Leer content_block.name de un evento delta en lugar de content_block_start - los deltas no llevan el nombre de la herramienta, solo content_block_start lo hace. Solución: capturar name e id cuando el bloque comienza, almacenarlos junto con el búfer.
Asumir que una llamada a herramienta sin argumentos aún transmite al menos un delta - una herramienta con un esquema de entrada vacío puede producir cero eventos input_json_delta. Solución: establecer el búfer en "{}" por defecto antes de analizar si no llegaron deltas.
Olvidar repetir el id de la herramienta como tool_use_id en el tool_result - Claude no puede hacer coincidir su resultado con la llamada correcta sin él. Solución: llevar content_block.id a través de su acumulador al resultado que envía de vuelta.
Intentar "renderizar parcialmente" los argumentos de la herramienta a un usuario a medida que se transmiten - a diferencia del texto, mostrar una cadena JSON a medio formar rara vez es útil y a menudo confuso. Solución: mostrar un indicador genérico "llamando a tool_name..." en lugar del JSON parcial sin procesar.

Alternativas

Alternativa	Usar cuándo	No usar cuándo
Uso de herramientas sin transmisión (`client.messages.create`)	La llamada completa a la herramienta, no el proceso de llegar a ella, es todo lo que su aplicación necesita	El usuario está viendo la respuesta en vivo y la latencia de selección de herramientas es importante
Un analizador "JSON parcial" tolerante (por ejemplo, intentar cerrar llaves abiertas)	Realmente necesita mostrar argumentos parciales que se actualizan en vivo en una interfaz de usuario de depuración	Rutas de código de producción: un análisis parcial mal formado puede producir silenciosamente una entrada de herramienta incorrecta
Almacenar en búfer toda la lista de eventos y procesar después de que el flujo se cierre	Código más simple, sin necesidad de una interfaz de usuario en vivo	Desea actuar sobre una llamada a herramienta en el momento en que está lista, antes de que termine el resto de la respuesta

Preguntas frecuentes

¿Qué contiene realmente un solo evento `input_json_delta`?

Un campo de cadena partial_json que contiene el siguiente fragmento del JSON de argumentos de la herramienta, desde unos pocos caracteres hasta unas pocas docenas, sin garantía de que se alinee con ningún límite de sintaxis JSON.

¿Cuándo es seguro llamar a `json.loads()` en la cadena acumulada?

Solo después de que content_block_stop se active para el índice de ese bloque. Antes de eso, la cadena acumulada puede ser (y generalmente es) JSON incompleto.

¿Cómo sé qué herramienta se está llamando antes de que sus argumentos terminen de transmitirse?

content_block_start incluye content_block.name y content_block.id inmediatamente, antes de que lleguen los eventos input_json_delta; capture esos primero.

¿Qué sucede si se llaman dos herramientas en la misma respuesta?

Se transmiten dos bloques de contenido tool_use separados, cada uno con su propio index. Mantenga un búfer por índice para que sus fragmentos JSON no se entrelacen en una cadena corrupta.

¿Puedo mostrar argumentos "escribiendo" en vivo al usuario como lo haría con texto?

No de forma útil, una cadena JSON a medio formar no significa nada para un lector. Muestre un indicador estático "llamando a tool_name..." en su lugar y revele los argumentos una vez analizados.

¿Qué pasa si `json.loads()` genera un error incluso después de `content_block_stop`?

Eso indica un error en su lógica de acumulación (por ejemplo, falta un delta, orden incorrecto o mezcla de búferes entre índices) en lugar de un problema de la API de Claude; la API garantiza que la cadena acumulada completa es JSON válido al cerrar el bloque.

¿Necesito el `tool_use_id` para algo más que el registro?

Sí, debe repetirse exactamente en el bloque de contenido tool_result que envíe en el siguiente turno, para que Claude pueda hacer coincidir su resultado con la llamada que realizó.

¿Es esto diferente de analizar JSON de una llamada a herramienta sin transmisión?

El resultado final (un diccionario de Python de json.loads) es idéntico. La diferencia radica solo en cómo se llega a él: la no transmisión le da la cadena JSON completa de una sola vez; la transmisión requiere acumularla primero.

¿Qué pasa si una herramienta no tiene argumentos requeridos?

Su bloque tool_use puede producir cero eventos input_json_delta. Establezca su búfer en un objeto JSON vacío ("{}") por defecto antes de analizar para no llamar a json.loads(""), lo que genera un error.

¿Cómo interactúa esto con el bucle más amplio de uso de herramientas agénticas?

Una vez que tiene la input analizada, la ejecución y el turno de tool_result de seguimiento funcionan exactamente igual que en el bucle de uso de herramientas sin transmisión; consulte Combinación de transmisión con el bucle de uso de herramientas para continuar la conversación.

Relacionado

Cómo los eventos enviados por el servidor impulsan las respuestas de transmisión de Claude - dónde encaja input_json_delta en el modelo general de eventos.
Combinación de transmisión con el bucle de uso de herramientas - qué hacer con una llamada a herramienta analizada una vez que la tiene.
Referencia de tipos de eventos de transmisión - la referencia completa de eventos/campos.
Definición de esquemas de herramientas con nombre, descripción y esquema de entrada - cómo se estructura el parámetro tools utilizado anteriormente.
Manejo del uso de herramientas de transmisión en flujos de trabajo asíncronos de Python - el mismo patrón en código asyncio.

Versiones de la pila: Escrito para la línea de modelos de Claude actual a partir de ~junio de 2026: Claude Fable 5, Claude Opus 4.8, Claude Sonnet 5 (el predeterminado) y Claude Haiku 4.5, y el SDK oficial de Python de anthropic (última versión 0.x). Los nombres de los modelos, las versiones del SDK y los precios cambian rápidamente; verifique los detalles actuales en platform.claude.com/docs antes de confiar en ellos.