Transmisión en tiempo real
Eventos enviados por el servidor en ambos formatos: sus estructuras difieren y, si los confundes, parecerá que no llega nada.
Configura stream en true y la respuesta pasa a ser text/event-stream. Ambos endpoints transmiten en streaming, pero emiten eventos con estructuras diferentes.
Formato de Anthropic
curl -N https://aiprimetech.io/v1/messages \
-H "x-api-key: $CLAUDEAPIKEY" \
-H "anthropic-version: 2023-06-01" \
-H "content-type: application/json" \
-d '{"model":"claude-sonnet-4-6","max_tokens":512,"stream":true,
"messages":[{"role":"user","content":"Count to five"}]}'
event: message_start
data: {"type":"message_start","message":{"id":"msg_...","usage":{"input_tokens":9}}}
event: content_block_delta
data: {"type":"content_block_delta","index":0,"delta":{"type":"text_delta","text":"One"}}
event: message_delta
data: {"type":"message_delta","delta":{"stop_reason":"end_turn"},"usage":{"output_tokens":14}}
event: message_stop
data: {"type":"message_stop"}
El texto llega en content_block_delta eventos en delta.text. Los recuentos finales de tokens de salida llegan en message_delta, no message_start.
Formato de OpenAI
data: {"choices":[{"delta":{"content":"One"},"index":0}]}
data: {"choices":[{"delta":{},"finish_reason":"stop","index":0}]}
data: [DONE]
El texto llega en choices[0].delta.content, y el flujo termina con un marcador literal data: [DONE] que no es JSON.
Esta discrepancia produce una respuesta vacía sin ningún error: un cliente que analiza
choices[].delta.content en un flujo de Messages no encuentra nada, siempre, e informa de que la operación se ha completado correctamente. Si el streaming parece funcionar pero no devuelve nada, comprueba qué estructura estás analizando.Python, SDK de Anthropic
from anthropic import Anthropic
client = Anthropic(api_key=class="s">"sk-your-key", base_url=class="s">"https://aiprimetech.io")
with client.messages.stream(
model=class="s">"claude-sonnet-4-6",
max_tokens=512,
messages=[{class="s">"role": class="s">"user", class="s">"content": class="s">"Count to five"}],
) as stream:
for text in stream.text_stream:
print(text, end=class="s">"", flush=True)
Notas operativas
- Usa
curl -Nal hacer pruebas manuales; de lo contrario, curl almacena todo el flujo en un búfer y parece que no ocurre nada. - Desactiva el almacenamiento en búfer del proxy situado delante de tu propia aplicación (
proxy_buffering offen nginx) o los clientes recibirán la respuesta de golpe al final. - Si la conexión se interrumpe durante el streaming, los tokens ya generados se facturan igualmente.
- El streaming no cambia el precio, solo el momento en que llegan los bytes.