FastAPI WebSocket: endpoint, gerenciador de broadcast e testes | We.Inc
Crie um endpoint WebSocket no FastAPI, faça broadcast para todos os clientes com um connection manager, leia tokens da query string, trate desconexões e teste tudo com o TestClient.
No FastAPI, você declara uma rota de socket com @app.websocket("/ws") e uma função async que recebe um WebSocket. Chame await websocket.accept() e depois faça um loop com receive_text() e send_text(). Capture WebSocketDisconnect para limpar tudo quando um cliente sair. Instale antes o pacote websockets e teste com TestClient.websocket_connect().
Os métodos de WebSocket que você mais vai chamar
Chamada
O que faz
await websocket.accept()
Conclui o handshake; nada pode ser enviado antes dele
await websocket.receive_text()
Espera a próxima mensagem de texto
await websocket.receive_json()
Espera uma mensagem e a interpreta como JSON
await websocket.send_text(s)
Envia uma mensagem de texto
await websocket.send_json(obj)
Serializa um objeto e o envia
await websocket.close(code=1000)
Encerra a conexão com um código de fechamento
O gerador de endpoint abaixo escreve um main.py funcional a partir de algumas escolhas: o caminho da rota, se as mensagens são texto simples ou JSON, se todo cliente recebe todas as mensagens e se um token precisa chegar na query string. Um arquivo pytest correspondente vem junto. Cada opção corresponde a um padrão da documentação do FastAPI, reescrito aqui como um único arquivo executável.
Escolha as opções no gerador e copie o main.py para uma pasta vazia.
Instale as dependências: pip install "fastapi[standard]" websockets, e depois inicie o app com fastapi dev main.py.
Conecte pelo testador de WebSocket da We.Inc ou pelo console do navegador em ws://localhost:8000 mais o seu caminho.
Copie o arquivo de teste gerado e rode pytest para verificar o endpoint sem iniciar um servidor.
Envolva o loop de recebimento em try e except WebSocketDisconnect. Sem isso, cada cliente que fecha a aba deixa um stack trace nos seus logs.
Fechar um socket antes do accept() faz o servidor responder ao handshake com HTTP 403, então o cliente nunca abre. Aceite primeiro e depois feche com 1008 se quiser que o cliente receba um código de violação de política legível.
Uma lista de conexões em memória só conhece os clientes do processo atual. Rode vários workers e cada um faz broadcast para um subconjunto diferente.
Use receive_json e send_json quando as mensagens tiverem mais de um campo. Eles poupam uma chamada a json.loads e falham de forma clara com entradas malformadas.
Clientes no navegador não conseguem adicionar um cabeçalho Authorization a um socket, então leia as credenciais de um parâmetro de query ou de um cookie.
Perguntas frequentes
Por que o meu WebSocket no FastAPI retorna 404 ou não faz o upgrade?
Comece pela instalação: a documentação de WebSocket do FastAPI pede que você adicione o pacote websockets junto com o FastAPI. Depois confira o caminho, porque uma rota declarada com @app.websocket só responde a upgrades de socket, nunca a GETs HTTP comuns. Fechar o socket antes do accept() também aparece como um handshake recusado (HTTP 403).
Como faço broadcast de uma mensagem para todos os clientes?
Guarde os sockets aceitos em uma lista dentro de uma pequena classe gerenciadora, adicione na conexão, remova na desconexão e percorra a lista chamando send_text. O gerador acima produz exatamente isso quando você marca a opção de broadcast.
O broadcast funciona com vários workers do Uvicorn?
Não com uma lista em memória. Cada worker é um processo separado com a própria lista. A documentação do FastAPI indica o encode/broadcaster, que retransmite as mensagens via Redis ou PostgreSQL para que todo processo receba todas as mensagens.
Posso usar Depends e Query em uma rota WebSocket?
Sim. Endpoints WebSocket aceitam Depends, Query, Cookie, Header, Path e Security assim como as rotas HTTP, então uma dependência de autenticação pode ser compartilhada entre os dois.
Como testo um endpoint WebSocket?
Crie um TestClient para o app e abra o socket em um bloco with usando client.websocket_connect("/ws"). Dentro dele, send_text e receive_text conduzem a conversa, e sair do bloco fecha a conexão.
O que o loop do endpoint está fazendo
Um handler de socket é uma corrotina de longa duração. Ele aceita uma vez e depois fica parado em await receive_text() até a próxima mensagem, então um cliente ocioso quase não custa nada. Quando o cliente vai embora, o receive pendente lança WebSocketDisconnect, e esse é o momento de arrumar a casa.
Como o handler é async, os outros clientes continuam sendo atendidos enquanto um está esperando. Evite chamadas bloqueantes como time.sleep ou drivers de banco de dados síncronos dentro do loop; elas congelam todas as conexões do worker.
Autenticação antes do accept
Leia o token da query string como um parâmetro normal. Você então tem duas formas de recusá-lo. Fechar antes do accept() vira um HTTP 403 no handshake, que o navegador só informa como uma conexão que falhou. Aceitar e depois fechar com o código 1008 permite que o cliente leia o motivo e mostre um pedido de login em vez de tentar de novo para sempre.
O menor endpoint funcional
main.py: echo
Aceite e depois devolva cada mensagem até o cliente se desconectar.
Broadcast com um connection manager
Uma classe gerenciadora compartilhada por todas as conexões
Funciona para um único processo. Troque a lista por um relay com Redis antes de escalar horizontalmente.
Testando com o TestClient
Não precisa de servidor: o cliente conversa com o app no mesmo processo. Rode com pytest.
Gerador de endpoint WebSocket para FastAPI
Escolha como o endpoint deve se comportar e copie um main.py completo e um arquivo pytest que o verifica.
Formato da mensagem
Texto simples
JSON
Enviar cada mensagem para todos os clientes conectados
Exigir um token na query string (?token=...)
Rode: fastapi dev main.py, e depois conecte em ws://localhost:8000{v}