O módulo socketserver simplifica a criação de servidores de rede no Python. Em vez de repetir manualmente o ciclo de criar um socket, fazer bind, escutar conexões, aceitar clientes e instanciar handlers, você escolhe uma classe de servidor e implementa apenas o método que processa cada requisição.
Essa abstração é útil em serviços internos, ferramentas de teste, protocolos simples, agentes locais e protótipos. Ainda assim, um servidor de rede precisa de limites, timeouts, framing, concorrência e encerramento previsível. Este guia mostra como usar TCPServer, UDPServer, StreamRequestHandler, servidores com threads e os principais cuidados de produção.
Como o socketserver é organizado
O módulo fornece quatro classes concretas básicas: TCPServer, UDPServer, UnixStreamServer e UnixDatagramServer. As duas últimas usam sockets Unix e não estão disponíveis em todas as plataformas. Por padrão, os servidores processam uma requisição de cada vez.
Para atender clientes simultaneamente, existem os mixins ThreadingMixIn e ForkingMixIn, além de classes prontas como ThreadingTCPServer e ThreadingUDPServer. O modelo com fork depende de POSIX e cria um novo processo, enquanto o modelo com threads compartilha memória e exige sincronização para estado mutável.
Antes de avançar, vale revisar o guia de cliente TCP com sockets no Python. socketserver não altera as propriedades de TCP ou UDP; apenas organiza o código do servidor.
Primeiro servidor TCP com StreamRequestHandler
StreamRequestHandler cria os objetos rfile e wfile, permitindo ler e escrever como em arquivos binários. O exemplo abaixo usa uma linha terminada por \n como framing.
import socketserver
MAX_LINE = 10_000
class EchoHandler(socketserver.StreamRequestHandler):
def handle(self) -> None:
line = self.rfile.readline(MAX_LINE + 1)
if len(line) > MAX_LINE:
self.wfile.write(b"erro: mensagem muito grande\n")
return
if not line.endswith(b"\n"):
self.wfile.write(b"erro: mensagem incompleta\n")
return
response = line.rstrip(b"\r\n").upper() + b"\n"
self.wfile.write(response)
HOST, PORT = "127.0.0.1", 9000
with socketserver.TCPServer((HOST, PORT), EchoHandler) as server:
server.serve_forever()
O limite de leitura é indispensável. Sem ele, um cliente pode enviar dados indefinidamente sem completar a linha, fazendo o processo acumular memória. Também é importante definir a codificação quando o protocolo usa texto, em vez de chamar decode() sem tratamento de erro.
TCP é um fluxo, não um conjunto de mensagens
Uma chamada sendall() do cliente não corresponde necessariamente a uma chamada recv() no servidor. Os bytes podem chegar divididos ou agrupados. Por isso, o protocolo precisa definir framing: linha terminada por delimitador, tamanho prefixado, formato binário de tamanho fixo ou fechamento da conexão.
O artigo sobre struct e dados binários no Python mostra como criar cabeçalhos com tamanho explícito. Para protocolos textuais simples, uma linha limitada costuma ser suficiente.
Servidor concorrente com threads
TCPServer é síncrono. Enquanto um cliente está sendo atendido, os demais aguardam. Para conexões com I/O ou clientes lentos, use ThreadingTCPServer.
class SafeThreadingTCPServer(socketserver.ThreadingTCPServer):
allow_reuse_address = True
daemon_threads = False
block_on_close = True
with SafeThreadingTCPServer((HOST, PORT), EchoHandler) as server:
server.serve_forever()
daemon_threads=False faz o Python aguardar as threads ativas antes de sair. block_on_close=True mantém o encerramento previsível. Se você escolher threads daemon, documente que requisições podem ser interrompidas durante o desligamento.
O guia de threading no Python explica locks, condições de corrida e quando threads ajudam. O GIL não elimina race conditions em estruturas compartilhadas.
Estado compartilhado e locks
Cada conexão recebe uma nova instância do handler, mas todas podem acessar self.server. Ao compartilhar contadores ou caches, proteja atualizações compostas.
import threading
class CountingServer(socketserver.ThreadingTCPServer):
daemon_threads = False
def __init__(self, address, handler):
super().__init__(address, handler)
self.total_requests = 0
self.counter_lock = threading.Lock()
class CountingHandler(socketserver.StreamRequestHandler):
def handle(self) -> None:
with self.server.counter_lock:
self.server.total_requests += 1
current = self.server.total_requests
self.wfile.write(f"requisição {current}\n".encode("utf-8"))
Evite manter sessões importantes apenas em memória quando usar processos, múltiplas instâncias ou reinicializações. Prefira armazenamento externo apropriado quando o estado precisa sobreviver ou ser compartilhado.
Timeout por conexão
Clientes lentos podem manter threads ocupadas. Configure timeout no socket aceito no método setup().
class TimedHandler(socketserver.StreamRequestHandler):
timeout_seconds = 10
def setup(self) -> None:
super().setup()
self.request.settimeout(self.timeout_seconds)
def handle(self) -> None:
try:
line = self.rfile.readline(4097)
except TimeoutError:
return
if not line or len(line) > 4096:
return
self.wfile.write(b"ok\n")
O atributo server.timeout afeta handle_request(), não o loop de serve_forever() nem automaticamente cada cliente. Para timeouts de sessão, configure o socket da conexão.
Backpressure e fila de conexões
request_queue_size controla aproximadamente quantas conexões aguardam aceitação quando o servidor está ocupado. Aumentar o valor não substitui limites de concorrência; apenas desloca a espera.
class LimitedServer(socketserver.ThreadingTCPServer):
request_queue_size = 32
allow_reuse_address = True
ThreadingTCPServer cria uma thread por conexão e não impõe um pool fixo. Um atacante pode abrir muitas conexões e consumir threads. Em serviço exposto, combine firewall, proxy, limites do sistema operacional, timeouts e, quando necessário, uma arquitetura com pool ou event loop.
Filtrar clientes com verify_request()
O método verify_request() pode rejeitar uma requisição antes do handler.
import ipaddress
ALLOWED = ipaddress.ip_network("10.10.0.0/16")
class InternalServer(socketserver.ThreadingTCPServer):
def verify_request(self, request, client_address) -> bool:
client_ip = ipaddress.ip_address(client_address[0])
return client_ip in ALLOWED
Uma allowlist por IP é apenas uma camada. Endereços podem mudar, proxies alteram a origem observada e redes internas também podem ser comprometidas. Para autenticação forte, use TLS com certificados, HMAC ou credenciais do protocolo. O guia de ssl e TLS no Python cobre a proteção do canal.
Servidor UDP
UDP trabalha com datagramas independentes, que podem ser perdidos, duplicados ou chegar fora de ordem.
class UDPHandler(socketserver.BaseRequestHandler):
def handle(self) -> None:
data, udp_socket = self.request
if len(data) > 1024:
return
response = data.strip().upper()
udp_socket.sendto(response, self.client_address)
with socketserver.ThreadingUDPServer((HOST, 9001), UDPHandler) as server:
server.serve_forever()
Não confie no endereço de origem UDP como autenticação. Evite respostas muito maiores que as solicitações para não criar um vetor de amplificação. Protocolos importantes precisam de IDs, proteção contra replay e tratamento de perda.
Tratamento de erros
Por padrão, uma exceção no handler é impressa em stderr e o servidor continua. Sobrescreva handle_error() para integrar logs estruturados sem expor payloads.
import logging
logger = logging.getLogger(__name__)
class LoggedServer(socketserver.ThreadingTCPServer):
def handle_error(self, request, client_address) -> None:
logger.exception(
"falha ao processar cliente",
extra={"client_ip": client_address[0]},
)
O guia de logging no Python ajuda a organizar níveis, handlers e dados de contexto. Não registre segredos nem mensagens completas por padrão.
Encerramento correto
serve_forever() para somente após uma chamada a shutdown(). Essa chamada precisa vir de outra thread; chamá-la dentro da mesma thread do loop causa deadlock.
import threading
server = SafeThreadingTCPServer((HOST, PORT), EchoHandler)
thread = threading.Thread(target=server.serve_forever)
thread.start()
try:
thread.join()
except KeyboardInterrupt:
server.shutdown()
server.server_close()
thread.join()
Em uma aplicação real, trate sinais no processo principal, pare de aceitar novas conexões, aguarde o trabalho em andamento até um prazo e então libere recursos.
IPv6
Para escutar em IPv6, ajuste address_family.
import socket
class IPv6TCPServer(socketserver.ThreadingTCPServer):
address_family = socket.AF_INET6
Teste dual stack em cada sistema, pois o comportamento de IPv4 mapeado pode variar. Valide endereços com o módulo ipaddress quando aplicar regras.
Quando usar socketserver?
Use o módulo para protocolos pequenos, ferramentas internas, mocks, servidores locais e aprendizado. Para HTTP público, prefira um framework e servidor WSGI/ASGI. Para milhares de conexões duradouras, considere asyncio ou uma arquitetura orientada a eventos. Para lógica pesada de CPU, filas externas e processos controlados oferecem limites melhores que uma thread por conexão.
Erros comuns
Os problemas mais frequentes são ler sem limite, assumir que um recv() contém uma mensagem inteira, usar o servidor síncrono com clientes lentos, compartilhar estado sem lock, criar threads ilimitadas, omitir timeout, confiar apenas no IP, chamar shutdown() na thread errada e expor um servidor de demonstração à internet.
Boas práticas
Defina framing, tamanho máximo e timeout. Escolha conscientemente entre síncrono, threads, processos ou event loop. Proteja estado compartilhado. Use TLS e autenticação quando necessário. Implemente logs sem payload sensível, métricas de conexões e encerramento testável. Coloque serviços públicos atrás de controles de rede e limites adicionais.
Conclusão
socketserver reduz bastante o código necessário para criar servidores TCP e UDP, mas não remove as responsabilidades de um protocolo de rede. Limites, framing, concorrência, autenticação e desligamento continuam essenciais. Para serviços pequenos e controlados, o módulo oferece uma estrutura clara; para cargas públicas e complexas, use abstrações com pools, event loops e infraestrutura de produção.
Consulte a documentação oficial de socketserver e o RFC 9293 sobre TCP.
