Transmissão
Eventos enviados pelo servidor em ambos os formatos — as estruturas dos eventos diferem, e confundi-las faz parecer que não há resposta.
Defina stream como true e a resposta passa a ser text/event-stream. Ambos os endpoints oferecem streaming, mas emitem estruturas de eventos diferentes.
Formato 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"}
O texto chega em content_block_delta eventos em delta.text. As contagens finais de tokens de saída chegam em message_delta, não message_start.
Formato OpenAI
data: {"choices":[{"delta":{"content":"One"},"index":0}]}
data: {"choices":[{"delta":{},"finish_reason":"stop","index":0}]}
data: [DONE]
O texto chega em choices[0].delta.content, e o fluxo termina com um marcador literal data: [DONE] que não é JSON.
Essa incompatibilidade produz uma resposta vazia sem erro: um cliente que interpreta
choices[].delta.content em um fluxo de Messages não encontra nada, em nenhuma tentativa, e informa sucesso. Se o streaming parecer funcionar, mas não retornar nada, verifique qual estrutura você está interpretando.Python, SDK da 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 operacionais
- Use
curl -Nao testar manualmente, ou o curl armazenará todo o fluxo em buffer e parecerá que nada está acontecendo. - Desative o buffering do proxy à frente do seu próprio aplicativo (
proxy_buffering offno nginx), ou os clientes receberão a resposta inteira de uma só vez ao final. - Se a conexão cair durante o streaming, os tokens já gerados ainda serão cobrados.
- O streaming não altera o preço — apenas o momento em que os bytes chegam.