Streaming

POST /chats/{chat_id}/messages/stream devuelve text/event-stream en vez de esperar al final. Sirve para dos cosas distintas: mostrar el texto mientras se genera, y poder contestarle al turno cuando pide permiso.

#El formato

Cada evento son dos líneas y una en blanco. data es siempre un JSON en una sola línea.

event: message.delta
data: {"text":"El pedido 8842","replace":false}

El stream siempre empieza con message.start y siempre termina con exactamente uno de message.completed, message.cancelled o message.error. Un fallo del proveedor seguido de un error es un terminal, no dos. Y el terminal sale recién cuando el resultado quedó guardado: si leíste message.completed, GET /chats/{chat_id}/messages ya lo devuelve.

#Los deltas son sufijos

message.delta trae el texto nuevo, no el acumulado. Concatenás lo que recibís y tenés la respuesta. Es la diferencia con el stream interno de la plataforma, que emite el acumulado y obligaría a reemplazar en cada evento.

La excepción es replace: true, que aparece cuando la respuesta se reescribe —pasa en un failover de proveedor—. Ahí text es el texto completo y hay que descartar lo acumulado hasta ese momento.

let answer = "";

function onDelta({ text, replace }) {
  // replace es raro, pero ignorarlo deja el texto duplicado tras un failover.
  answer = replace ? text : answer + text;
}

#Keepalive y desconexión

Cada 15 segundos de silencio se emite un ping con un timestamp. Sirve para mantener viva la conexión a través de proxies; ignoralo en tu lógica, pero úsalo como señal de que el turno sigue vivo.

El turno también tiene un plazo máximo de 30 minutos. Al vencer recibís message.error con code: "turn_timeout" y retryable: false.

#Un cliente completo

import json
import uuid

import requests


def stream_turn(chat_id: str, content: str, token: str):
    """Envía un turno y produce (evento, payload) hasta el terminal."""
    with requests.post(
        f"https://api.niucore.com/api/v1/chats/{chat_id}/messages/stream",
        headers={
            "Authorization": f"Bearer {token}",
            "Accept": "text/event-stream",
            "Idempotency-Key": str(uuid.uuid4()),
        },
        json={"content": content, "permission_mode": "manual"},
        stream=True,
        # Sin timeout de lectura: un turno legítimo puede callarse hasta 15 s
        # entre pings, y mucho más entre un paso y el siguiente.
        timeout=(10, None),
    ) as response:
        response.raise_for_status()

        event = None
        for line in response.iter_lines(decode_unicode=True):
            if not line:
                continue
            if line.startswith("event:"):
                event = line[6:].strip()
            elif line.startswith("data:"):
                yield event, json.loads(line[5:].strip())


answer = ""
for event, payload in stream_turn(chat_id, "¿Qué pasó con el pedido 8842?", token):
    if event == "ping":
        continue
    if event == "message.delta":
        answer = payload["text"] if payload["replace"] else answer + payload["text"]
    elif event == "message.action_required":
        handle_interaction(payload)          # ver la guía de interacciones
    elif event == "message.completed":
        print(payload["usage"])
    elif event in ("message.cancelled", "message.error"):
        print(event, payload)

#Proxies y buffering

La respuesta viaja con Cache-Control: no-cache, no-transform y X-Accel-Buffering: no. Si igual ves el stream llegar de golpe al final, el culpable suele estar en tu lado:

• Un cliente HTTP que no expone el cuerpo hasta completarlo (stream=True en requests, no leer response.text).
• Un proxy inverso que comprime al vuelo. no-transform lo pide, pero no todos lo respetan.
• Un serverless con respuesta bufferizada: muchos entornos no soportan streaming de salida.