WebSockets com Python — 2026 | Python Brasil
Aprenda a criar aplicacoes em tempo real com WebSockets e Python. Tutorial com chat, notificacoes e integracao com FastAPI e websockets.
6 min de leitura
Equipe Python Brasil
WebSockets permitem comunicacao bidirecional em tempo real entre cliente e servidor. Diferente do HTTP tradicional, onde o cliente precisa fazer uma requisicao para receber dados, com WebSockets o servidor pode enviar mensagens a qualquer momento. Neste guia, a gente vai construir aplicacoes em tempo real com Python.
Como WebSockets Funcionam
No protocolo HTTP normal, cada requisicao abre e fecha uma conexao. WebSockets estabelecem uma conexao persistente atraves de um handshake inicial HTTP, e depois manteem um canal aberto para troca de mensagens em ambas as direcoes.
Isso e ideal para aplicacoes como chats, dashboards ao vivo, jogos multiplayer e notificacoes em tempo real.
Servidor WebSocket Basico
Vamos comecar com a biblioteca websockets:
pip install websockets
Servidor simples:
# servidor.py
import asyncio
import websockets
import json
from datetime import datetime
async def handler(websocket):
"""Trata conexoes WebSocket."""
print(f"Cliente conectado: {websocket.remote_address}")
try:
async for mensagem in websocket:
dados = json.loads(mensagem)
print(f"Recebido: {dados}")
# Processar e responder
resposta = {
"tipo": "resposta",
"mensagem": f"Voce disse: {dados.get('mensagem', '')}",
"timestamp": datetime.now().isoformat()
}
await websocket.send(json.dumps(resposta))
except websockets.exceptions.ConnectionClosed:
print(f"Cliente desconectado: {websocket.remote_address}")
async def main():
async with websockets.serve(handler, "localhost", 8765):
print("Servidor WebSocket rodando em ws://localhost:8765")
await asyncio.Future() # Roda para sempre
if __name__ == "__main__":
asyncio.run(main())
Cliente simples:
# cliente.py
import asyncio
import websockets
import json
async def conectar():
"""Conecta ao servidor WebSocket."""
uri = "ws://localhost:8765"
async with websockets.connect(uri) as ws:
# Enviar mensagem
mensagem = {"mensagem": "Ola do cliente Python!"}
await ws.send(json.dumps(mensagem))
print(f"Enviado: {mensagem}")
# Receber resposta
resposta = await ws.recv()
dados = json.loads(resposta)
print(f"Recebido: {dados}")
if __name__ == "__main__":
asyncio.run(conectar())
Chat em Tempo Real
Vamos construir um servidor de chat completo:
# chat_server.py
import asyncio
import websockets
import json
from datetime import datetime
# Conjunto de clientes conectados
clientes = {}
async def registrar(websocket, nome):
"""Registra um novo cliente."""
clientes[websocket] = nome
await broadcast({
"tipo": "sistema",
"mensagem": f"{nome} entrou no chat",
"timestamp": datetime.now().isoformat(),
"online": list(clientes.values())
})
async def desregistrar(websocket):
"""Remove um cliente desconectado."""
nome = clientes.pop(websocket, "Desconhecido")
await broadcast({
"tipo": "sistema",
"mensagem": f"{nome} saiu do chat",
"timestamp": datetime.now().isoformat(),
"online": list(clientes.values())
})
async def broadcast(mensagem):
"""Envia mensagem para todos os clientes conectados."""
if clientes:
msg_json = json.dumps(mensagem, ensure_ascii=False)
await asyncio.gather(
*[ws.send(msg_json) for ws in clientes.keys()],
return_exceptions=True
)
async def handler(websocket):
"""Trata conexoes do chat."""
nome = None
try:
async for msg_raw in websocket:
dados = json.loads(msg_raw)
if dados["tipo"] == "entrar":
nome = dados["nome"]
await registrar(websocket, nome)
elif dados["tipo"] == "mensagem":
await broadcast({
"tipo": "mensagem",
"autor": clientes.get(websocket, "Anonimo"),
"mensagem": dados["mensagem"],
"timestamp": datetime.now().isoformat()
})
except websockets.exceptions.ConnectionClosed:
pass
finally:
if websocket in clientes:
await desregistrar(websocket)
async def main():
async with websockets.serve(handler, "0.0.0.0", 8765):
print("Servidor de chat rodando em ws://0.0.0.0:8765")
await asyncio.Future()
if __name__ == "__main__":
asyncio.run(main())
Cliente de chat interativo:
# chat_client.py
import asyncio
import websockets
import json
import sys
async def receber_mensagens(ws):
"""Recebe e exibe mensagens do servidor."""
async for msg_raw in ws:
dados = json.loads(msg_raw)
if dados["tipo"] == "sistema":
print(f"\n[SISTEMA] {dados['mensagem']}")
print(f"Online: {', '.join(dados['online'])}")
elif dados["tipo"] == "mensagem":
print(f"\n[{dados['autor']}] {dados['mensagem']}")
async def enviar_mensagens(ws):
"""Le input do usuario e envia mensagens."""
loop = asyncio.get_event_loop()
while True:
texto = await loop.run_in_executor(None, input)
if texto.lower() == "/sair":
break
mensagem = {"tipo": "mensagem", "mensagem": texto}
await ws.send(json.dumps(mensagem))
async def main():
nome = input("Seu nome: ")
uri = "ws://localhost:8765"
async with websockets.connect(uri) as ws:
# Registrar no chat
await ws.send(json.dumps({"tipo": "entrar", "nome": nome}))
# Executar envio e recebimento em paralelo
receber = asyncio.create_task(receber_mensagens(ws))
enviar = asyncio.create_task(enviar_mensagens(ws))
done, pending = await asyncio.wait(
[receber, enviar],
return_when=asyncio.FIRST_COMPLETED
)
for task in pending:
task.cancel()
if __name__ == "__main__":
asyncio.run(main())
WebSockets com FastAPI
FastAPI tem suporte nativo a WebSockets, facilitando a integracao com APIs REST:
# main.py
from fastapi import FastAPI, WebSocket, WebSocketDisconnect
from fastapi.responses import HTMLResponse
import json
from datetime import datetime
app = FastAPI()
class GerenciadorConexoes:
"""Gerencia conexoes WebSocket ativas."""
def __init__(self):
self.conexoes: dict[str, WebSocket] = {}
async def conectar(self, websocket: WebSocket, usuario_id: str):
await websocket.accept()
self.conexoes[usuario_id] = websocket
def desconectar(self, usuario_id: str):
self.conexoes.pop(usuario_id, None)
async def enviar_para(self, usuario_id: str, mensagem: dict):
ws = self.conexoes.get(usuario_id)
if ws:
await ws.send_json(mensagem)
async def broadcast(self, mensagem: dict, excluir: str = None):
for uid, ws in self.conexoes.items():
if uid != excluir:
try:
await ws.send_json(mensagem)
except Exception:
pass
gerenciador = GerenciadorConexoes()
@app.websocket("/ws/{usuario_id}")
async def websocket_endpoint(websocket: WebSocket, usuario_id: str):
await gerenciador.conectar(websocket, usuario_id)
await gerenciador.broadcast({
"tipo": "sistema",
"mensagem": f"{usuario_id} conectou",
"timestamp": datetime.now().isoformat()
})
try:
while True:
dados = await websocket.receive_json()
if dados.get("tipo") == "mensagem":
await gerenciador.broadcast({
"tipo": "mensagem",
"autor": usuario_id,
"mensagem": dados["mensagem"],
"timestamp": datetime.now().isoformat()
})
elif dados.get("tipo") == "privada":
await gerenciador.enviar_para(
dados["destinatario"],
{
"tipo": "privada",
"autor": usuario_id,
"mensagem": dados["mensagem"],
"timestamp": datetime.now().isoformat()
}
)
except WebSocketDisconnect:
gerenciador.desconectar(usuario_id)
await gerenciador.broadcast({
"tipo": "sistema",
"mensagem": f"{usuario_id} desconectou"
})
@app.get("/")
async def index():
return {"mensagem": "Conecte via WebSocket em /ws/{usuario_id}"}
Notificacoes em Tempo Real
Um caso de uso pratico e enviar notificacoes para usuarios especificos:
from fastapi import FastAPI, WebSocket, WebSocketDisconnect
from datetime import datetime
app = FastAPI()
conexoes_notificacao: dict[str, WebSocket] = {}
@app.websocket("/notificacoes/{usuario_id}")
async def notificacoes_ws(websocket: WebSocket, usuario_id: str):
"""WebSocket para receber notificacoes."""
await websocket.accept()
conexoes_notificacao[usuario_id] = websocket
try:
while True:
await websocket.receive_text() # Manter conexao ativa
except WebSocketDisconnect:
conexoes_notificacao.pop(usuario_id, None)
async def enviar_notificacao(usuario_id: str, titulo: str, mensagem: str):
"""Funcao utilitaria para enviar notificacao."""
ws = conexoes_notificacao.get(usuario_id)
if ws:
await ws.send_json({
"tipo": "notificacao",
"titulo": titulo,
"mensagem": mensagem,
"timestamp": datetime.now().isoformat()
})
# Endpoint REST que dispara notificacao via WebSocket
@app.post("/pedidos/{pedido_id}/aprovar")
async def aprovar_pedido(pedido_id: int):
usuario_id = "usuario_123" # Obtido do banco de dados
await enviar_notificacao(
usuario_id,
"Pedido Aprovado",
f"Seu pedido #{pedido_id} foi aprovado!"
)
return {"status": "aprovado"}
Reconexao Automatica no Cliente
Clientes devem reconectar automaticamente se a conexao cair:
import asyncio
import websockets
import json
async def conectar_com_reconexao(uri, nome):
"""Conecta com reconexao automatica."""
while True:
try:
async with websockets.connect(uri) as ws:
print(f"Conectado a {uri}")
await ws.send(json.dumps({"tipo": "entrar", "nome": nome}))
async for mensagem in ws:
dados = json.loads(mensagem)
print(f"Recebido: {dados}")
except (websockets.exceptions.ConnectionClosed, ConnectionRefusedError):
print("Conexao perdida. Reconectando em 3 segundos...")
await asyncio.sleep(3)
except Exception as e:
print(f"Erro: {e}. Reconectando em 5 segundos...")
await asyncio.sleep(5)
Boas Praticas
Ao trabalhar com WebSockets em Python, siga estas recomendacoes:
- Implemente heartbeat/ping-pong para detectar conexoes mortas
- Use JSON como formato padrao para mensagens estruturadas
- Adicione autenticacao no handshake inicial via query parameters ou headers
- Trate desconexoes de forma graciosa com cleanup adequado
- Limite o numero de conexoes simultaneas por usuario
- Use salas ou canais para segmentar mensagens por contexto
- Implemente reconexao automatica no cliente
Conclusao
WebSockets com Python abrem portas para aplicacoes em tempo real poderosas. Seja um chat, dashboard ao vivo ou sistema de notificacoes, as bibliotecas websockets e FastAPI oferecem tudo que voce precisa. Comece com um servidor simples, evolua para o gerenciador de conexoes e implemente features avancadas como mensagens privadas e reconexao automatica conforme a necessidade do seu projeto.
E
Equipe Python Brasil
Contribuidor do Python Brasil — Aprenda Python em Português