Streaming do Claude no Laravel com SSE: como mostrar a resposta enquanto é gerada

Como repassar o stream da API do Claude até o navegador no Laravel com SSE: parser de eventos, flush, Nginx, desconexão do cliente, erro no meio do stream e registro de tokens.

Para fazer streaming da resposta do Claude no Laravel, você chama a Messages API com "stream": true e lê o corpo da resposta aos poucos com o Http:: em modo stream. Cada text_delta vira um evento SSE devolvido por um response()->stream(). No navegador, um fetch lê esse corpo incrementalmente. O código é curto. O trabalho de verdade está nas bordas: buffers do Nginx e do PHP, cliente que fecha a aba, erro no meio da geração e contabilizar tokens de uma resposta que talvez nunca termine.

No WhatsApp, streaming não faz diferença, porque a mensagem sai inteira. Num assistente dentro de um painel web, faz muita. Uma resposta de 20 segundos com a tela parada parece travamento. A mesma resposta aparecendo palavra por palavra parece rápida. Abaixo está o caminho que eu sigo, do token da API até a tela.

O que é SSE e por que usar em vez de WebSocket?

SSE (Server-Sent Events) é um formato de texto sobre HTTP comum. O servidor mantém a resposta aberta e vai escrevendo blocos event: / data:, separados por uma linha em branco. A própria API da Anthropic usa esse formato para o streaming, então repassar SSE para o navegador é o caminho de menor atrito.

WebSocket é bidirecional e exige um servidor dedicado, como o Laravel Reverb. Para "usuário manda uma pergunta, servidor devolve uma resposta longa", essa bidirecionalidade não serve para nada. Existem três desenhos possíveis:

Abordagem	Como funciona	Quando usar
Sem streaming	Job na fila chama o Claude e notifica no fim	Tarefas em background, WhatsApp, extrações
SSE direto no controller	A request HTTP fica aberta repassando os deltas	Chat em painel web, poucas conexões simultâneas
Job + Reverb	Job consome o stream e faz broadcast dos deltas	Muitos usuários simultâneos, ou quando o PHP-FPM é gargalo

Neste texto eu sigo com SSE direto, que resolve a maioria dos casos. O trade-off dele aparece mais adiante.

Quais eventos a API do Claude envia no streaming?

Com stream: true, a Messages API emite esta sequência de eventos:

message_start: traz o objeto da mensagem, incluindo usage.input_tokens.
content_block_start: abre um bloco de conteúdo (text ou tool_use).
content_block_delta: pedaços do bloco. Para texto, delta.type é text_delta com o campo text. Para tool use, é input_json_delta com partial_json.
content_block_stop: fecha o bloco.
message_delta: traz o stop_reason e o usage.output_tokens acumulado.
message_stop: fim da mensagem.

No meio disso podem chegar eventos ping, que servem só para manter a conexão viva. Pode chegar também um evento error, como overloaded_error, mesmo depois de a resposta ter começado com status 200. Esse último detalhe muda o tratamento de erro, e eu volto nele adiante.

Como ler o stream da API com o Http do Laravel?

O Http:: do Laravel aceita a opção stream do Guzzle. Com ela, o corpo não é baixado de uma vez e você lê pela PSR stream via toPsrResponse()->getBody(). Encapsulo isso num generator que devolve eventos já parseados:

namespace App\Ai;

use Illuminate\Support\Facades\Http;
use RuntimeException;

final class ClaudeStream
{
    /** @return \Generator<int, array{event: string, data: array}> */
    public function stream(array $payload): \Generator
    {
        $response = Http::withHeaders([
                'x-api-key' => config('services.anthropic.key'),
                'anthropic-version' => '2023-06-01',
            ])
            ->connectTimeout(5)
            ->timeout(180)
            ->withOptions(['stream' => true, 'read_timeout' => 30])
            ->post('https://api.anthropic.com/v1/messages', [...$payload, 'stream' => true]);

        if ($response->failed()) {
            throw new RuntimeException('Claude HTTP '.$response->status());
        }

        $body = $response->toPsrResponse()->getBody();
        $buffer = '';

        while (! $body->eof()) {
            $buffer .= str_replace("\r\n", "\n", $body->read(1024));

            while (($pos = strpos($buffer, "\n\n")) !== false) {
                $raw = substr($buffer, 0, $pos);
                $buffer = substr($buffer, $pos + 2);

                $event = null;
                $data = '';
                foreach (explode("\n", $raw) as $line) {
                    if (str_starts_with($line, 'event:')) {
                        $event = trim(substr($line, 6));
                    } elseif (str_starts_with($line, 'data:')) {
                        $data .= trim(substr($line, 5));
                    }
                }

                if ($event !== null && $data !== '') {
                    yield ['event' => $event, 'data' => json_decode($data, true)];
                }
            }
        }
    }
}

Três detalhes que costumam passar batido:

Timeout padrão de 30 segundos. O cliente HTTP do Laravel vem com timeout de 30 segundos, e uma resposta longa ultrapassa isso fácil. Eu defino um teto total alto com timeout() e uso o read_timeout para outra função: detectar silêncio entre um chunk e outro. Como a API manda ping, 30 segundos sem nenhum byte significa conexão morta.
Um read() não corresponde a um evento. Um único read() pode trazer meio evento ou três eventos de uma vez. Por isso existe o buffer e o corte em \n\n. Fazer json_decode direto no chunk funciona no teste local e quebra em produção.
$response->failed() só cobre erro antes do stream. Ele pega 401, 429 e 529 devolvidos de cara. Erro depois do primeiro byte chega como evento error dentro do corpo com status 200.

Como devolver SSE para o navegador no Laravel?

No controller, uso response()->stream() com os headers certos e faço flush a cada evento:

public function __invoke(Request $request, Conversation $conversation, ClaudeStream $claude)
{
    $payload = [
        'model' => config('services.anthropic.model'),
        'max_tokens' => 2048,
        'system' => $conversation->systemPrompt(),
        'messages' => $conversation->messagesFor($request->string('message')),
    ];

    return response()->stream(function () use ($claude, $payload, $conversation) {
        ignore_user_abort(true);

        $text = '';
        $usage = ['input' => 0, 'output' => 0];
        $status = 'interrupted';

        $send = function (string $event, array $data): void {
            echo "event: {$event}\n";
            echo 'data: '.json_encode($data)."\n\n";
            if (ob_get_level() > 0) {
                ob_flush();
            }
            flush();
        };

        try {
            foreach ($claude->stream($payload) as $e) {
                switch ($e['event']) {
                    case 'message_start':
                        $usage['input'] = $e['data']['message']['usage']['input_tokens'] ?? 0;
                        break;
                    case 'content_block_delta':
                        if (($e['data']['delta']['type'] ?? null) === 'text_delta') {
                            $text .= $e['data']['delta']['text'];
                            $send('delta', ['text' => $e['data']['delta']['text']]);
                        }
                        break;
                    case 'message_delta':
                        $usage['output'] = $e['data']['usage']['output_tokens'] ?? $usage['output'];
                        break;
                    case 'message_stop':
                        $status = 'complete';
                        break;
                    case 'error':
                        $status = 'error';
                        $send('error', ['message' => 'A resposta foi interrompida. Tente de novo.']);
                        break 2;
                }

                if (connection_aborted()) {
                    break;
                }
            }

            if ($status === 'complete') {
                $send('done', []);
            }
        } catch (\Throwable $ex) {
            $status = 'error';
            report($ex);
            $send('error', ['message' => 'Não consegui responder agora.']);
        } finally {
            $conversation->messages()->create([
                'role' => 'assistant',
                'content' => $text,
                'status' => $status,
                'input_tokens' => $usage['input'],
                'output_tokens' => $usage['output'],
            ]);
        }
    }, 200, [
        'Content-Type' => 'text/event-stream',
        'Cache-Control' => 'no-cache',
        'X-Accel-Buffering' => 'no',
    ]);
}

Os dois detalhes que mais importam aqui são o ignore_user_abort(true) e o finally. Sem ignore_user_abort, o PHP encerra o script quando tenta escrever para um cliente que já foi embora. O finally não roda, e você perde tanto o texto gerado quanto o registro de tokens. Com ele, o script continua, o connection_aborted() passa a refletir a desconexão depois do próximo flush, e você sai do loop do seu jeito.

O que fazer quando o usuário fecha a aba no meio da resposta?

Você pode continuar consumindo o stream até o fim ou cortar. Eu corto. Ao sair do loop, o generator é descartado e a conexão com a Anthropic é fechada, o que interrompe a geração. Continuar lendo só faria sentido se a resposta completa tivesse valor sem ninguém olhando, e num chat isso quase nunca acontece.

O preço de cortar é que o message_delta com o output_tokens final nunca chega. O que fica registrado é o último valor visto, ou zero. Para controle de custo, eu marco essas mensagens como interrupted e estimo os tokens de saída pelo tamanho do texto acumulado. Se você impõe cota por cliente, como no post sobre como medir tokens e limitar o custo de IA por cliente no Laravel, essa estimativa precisa entrar na conta. Do contrário, quem fecha a aba no meio vira um buraco na medição.

Outra decisão é se a mensagem interrompida entra no histórico da próxima chamada. Eu incluo o texto parcial quando ele tem conteúdo útil e deixo de fora quando está vazio. Mandar ao modelo uma resposta pela metade sem aviso pode fazê-lo continuar de onde parou, o que confunde. Uma alternativa é anexar uma nota curta no histórico dizendo que a resposta anterior foi interrompida.

Como tratar erro no meio do streaming do Claude?

Esse é o ponto em que streaming é mais difícil que a chamada síncrona. Numa chamada normal, um 529 (API sobrecarregada) vira um retry com backoff e o usuário nem percebe. No stream, se o erro chega depois de o usuário ler três parágrafos, um retry transparente não existe: a nova resposta não vai repetir as mesmas palavras.

A regra que eu uso tem duas metades:

Erro antes do primeiro text_delta: pode tentar de novo, ou cair para outro provedor, porque o usuário ainda não viu nada. Aqui vale a mesma lógica de fallback de LLM quando a API do Claude cai. Só é preciso checar se já houve delta antes de decidir.
Erro depois do primeiro delta: não tento esconder. Mando um evento error para o front, salvo a mensagem como error com o texto parcial e deixo um botão "tentar de novo" que gera uma resposta nova do zero.

Na prática, eu guardo uma flag $started = true no primeiro delta e envolvo o foreach num retry que só se aplica enquanto $started for falso.

E quando o agente chama ferramentas durante o stream?

Se o payload tem tools, o modelo pode abrir um bloco tool_use no meio da resposta. O input da ferramenta chega fatiado em eventos input_json_delta, e cada partial_json é um fragmento de JSON que não é válido isoladamente. A regra é concatenar por índice do bloco e só fazer json_decode no content_block_stop:

case 'content_block_start':
    if ($e['data']['content_block']['type'] === 'tool_use') {
        $tools[$e['data']['index']] = [
            'id' => $e['data']['content_block']['id'],
            'name' => $e['data']['content_block']['name'],
            'json' => '',
        ];
    }
    break;

case 'content_block_delta':
    if ($e['data']['delta']['type'] === 'input_json_delta') {
        $tools[$e['data']['index']]['json'] .= $e['data']['delta']['partial_json'];
    }
    break;

Para o usuário, eu mando um evento status do tipo "consultando seus pedidos..." enquanto a ferramenta executa. Depois disso, abro um novo stream com o tool_result. O loop do agente continua igual ao síncrono, com os mesmos cuidados de stop_reason e limite de iterações que descrevi em erros no loop do agente de IA no Laravel. A única diferença é que cada volta do loop é um stream.

Como consumir o SSE no React com fetch?

O EventSource nativo só faz GET e não permite enviar corpo nem headers customizados. Como a pergunta do usuário vai no corpo, uso fetch e leio o ReadableStream:

const res = await fetch(`/conversations/${id}/stream`, {
  method: 'POST',
  headers: { 'Content-Type': 'application/json', 'X-CSRF-TOKEN': csrf },
  body: JSON.stringify({ message }),
});

const reader = res.body.getReader();
const decoder = new TextDecoder();
let buffer = '';

while (true) {
  const { value, done } = await reader.read();
  if (done) break;
  buffer += decoder.decode(value, { stream: true });

  let idx;
  while ((idx = buffer.indexOf('\n\n')) !== -1) {
    const raw = buffer.slice(0, idx);
    buffer = buffer.slice(idx + 2);
    const event = raw.match(/^event: (.+)$/m)?.[1];
    const data = JSON.parse(raw.match(/^data: (.+)$/m)?.[1] ?? '{}');
    if (event === 'delta') setText((t) => t + data.text);
    if (event === 'error') setError(data.message);
  }
}

O { stream: true } no TextDecoder evita que um caractere acentuado partido entre dois chunks vire lixo na tela. Em português isso aparece rápido.

Por que o streaming funciona local e chega tudo de uma vez em produção?

Quase sempre o culpado é um buffer no caminho. A checklist que eu passo:

Nginx: o header X-Accel-Buffering: no desliga o buffer do proxy para aquela resposta. Se o PHP-FPM fica atrás de fastcgi_pass, confira também fastcgi_buffering.
Compressão: gzip no Nginx para text/event-stream acumula bytes antes de enviar. Tire esse content type da lista do gzip_types.
PHP: output_buffering no php.ini segura a saída. O ob_flush() seguido de flush() resolve, mas só se você checar ob_get_level() antes, para não gerar notice.
CDN ou load balancer: alguns proxies na frente do servidor bufferizam ou têm timeout de ociosidade curto. Teste o endpoint com curl -N passando por eles, não só direto no servidor.
Middleware que mexe no corpo: qualquer middleware que lê ou reescreve o conteúdo da resposta não combina com StreamedResponse. Rotas de stream ficam num grupo sem esses middlewares.

Qual o custo de manter uma request aberta no PHP-FPM?

Esse é o trade-off do SSE direto. Cada resposta em streaming ocupa um worker do PHP-FPM durante toda a geração, e isso pode passar de 20 ou 30 segundos numa resposta longa com tool calls. Com pm.max_children dimensionado para requests de 200 ms, poucos usuários conversando ao mesmo tempo já esgotam o pool, e o resto do painel começa a enfileirar.

As saídas, em ordem de esforço:

Pool separado. Um pool do PHP-FPM só para as rotas de stream, roteado pelo Nginx por location. O chat lento deixa de competir com o resto do app. É a mesma ideia de bulkhead que uso nas filas.
Octane. Com Swoole ou FrankenPHP, o modelo de concorrência é outro. Só vale se o app já roda bem em Octane, porque estado vazando entre requests é um problema próprio.
Job + Reverb. O controller só despacha um job, o job consome o stream do Claude e faz broadcast dos deltas pelo Reverb. Nesse caso eu agrupo os deltas em janelas de 100 a 200 ms, porque um broadcast por token é desperdício. O worker da fila fica preso no lugar do FPM, mas fila é muito mais fácil de escalar e isolar.

Streaming não muda o preço por token: você paga o mesmo que na chamada síncrona. Se quiser simular quanto o chat vai custar por mês com e sem prompt caching, a calculadora de custo de IA faz essa conta por modelo.

Conclusão: quando vale a pena fazer streaming

Use streaming quando há um humano olhando para a tela esperando a resposta. Para todo o resto (WhatsApp, extração, conciliação, jobs), a chamada síncrona numa fila é mais simples, tem retry de verdade e devolve o usage completo.

Se for implementar, comece pelas bordas e não pelo parser. Garanta ignore_user_abort e finally para nunca perder o registro da mensagem. Separe o erro "antes do primeiro delta", que admite retry, do erro "depois", que deve ser mostrado ao usuário. Teste com curl -N passando por toda a infraestrutura de produção. E dimensione o pool do PHP-FPM pensando em conexões que ficam abertas por dezenas de segundos. O resto são uns 80 linhas de PHP que você escreve uma vez e reaproveita em todo assistente do produto.