El módulo socket de Python expone directamente la API de sockets del sistema operativo. Es la base sobre la que están construidas todas las bibliotecas de red de más alto nivel: http.server, asyncio, paramiko. Entender cómo funciona te permite depurar problemas de red, implementar protocolos personalizados y sacar el máximo partido a las abstracciones de nivel superior.
Servidor TCP básico
import socket
def servidor_tcp(host: str = '127.0.0.1', puerto: int = 9000):
# AF_INET = IPv4, SOCK_STREAM = TCP
with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as servidor:
servidor.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
servidor.bind((host, puerto))
servidor.listen(5) # cola de hasta 5 conexiones pendientes
print(f"Escuchando en {host}:{puerto}")
while True:
conn, direccion = servidor.accept()
with conn:
print(f"Conexión de {direccion}")
datos = conn.recv(1024)
if datos:
respuesta = datos.upper()
conn.sendall(respuesta)
# servidor_tcp() # descomentar para ejecutar
Cliente TCP
import socket
def cliente_tcp(mensaje: str, host: str = '127.0.0.1', puerto: int = 9000) -> str:
with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as cliente:
cliente.connect((host, puerto))
cliente.sendall(mensaje.encode('utf-8'))
datos = cliente.recv(1024)
return datos.decode('utf-8')
# respuesta = cliente_tcp("hola mundo")
# print(respuesta) # HOLA MUNDO
El error "Address already in use"
Si tu servidor se cierra sin liberar el socket, el sistema operativo mantiene el puerto en estado TIME_WAIT durante unos segundos. La solución es siempre usar SO_REUSEADDR:
servidor.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) servidor.bind((host, puerto))
UDP: sin conexión
UDP no establece conexión: los datagramas se envían directamente. No hay garantía de entrega ni de orden. Es apropiado para streaming de vídeo, DNS o protocolos donde la velocidad prima sobre la fiabilidad.
import socket
# Servidor UDP
def servidor_udp(host='127.0.0.1', puerto=9001):
with socket.socket(socket.AF_INET, socket.SOCK_DGRAM) as s:
s.bind((host, puerto))
print(f"UDP escuchando en {host}:{puerto}")
while True:
datos, direccion = s.recvfrom(1024)
print(f"De {direccion}: {datos.decode()}")
s.sendto(datos.upper(), direccion)
# Cliente UDP
def cliente_udp(mensaje: str, host='127.0.0.1', puerto=9001) -> str:
with socket.socket(socket.AF_INET, socket.SOCK_DGRAM) as s:
s.sendto(mensaje.encode(), (host, puerto))
datos, _ = s.recvfrom(1024)
return datos.decode()
select: multiplexar varias conexiones sin hilos
select.select() bloquea hasta que alguno de los sockets especificados está listo para leer, escribir o tiene un error. Es la base de los servidores de un solo hilo que atienden múltiples clientes:
import socket
import select
def servidor_multiplex(host='127.0.0.1', puerto=9002):
servidor = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
servidor.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
servidor.bind((host, puerto))
servidor.listen(10)
servidor.setblocking(False)
entradas = [servidor] # sockets a monitorizar
print(f"Servidor multiplex en {host}:{puerto}")
while entradas:
legibles, _, _ = select.select(entradas, [], [], timeout=1.0)
for s in legibles:
if s is servidor:
conn, addr = servidor.accept()
conn.setblocking(False)
entradas.append(conn)
print(f"Nueva conexión: {addr}")
else:
datos = s.recv(1024)
if datos:
s.sendall(datos.upper())
else:
entradas.remove(s)
s.close()
struct: serializar mensajes binarios
Los sockets transmiten bytes. Para enviar estructuras de datos con múltiples campos se usa struct, que empaqueta tipos Python en representaciones binarias de tamaño fijo:
import socket
import struct
# Protocolo: cabecera de 8 bytes (longitud del cuerpo como uint64 big-endian)
# + cuerpo de N bytes (UTF-8)
CABECERA = struct.Struct('!Q') # ! = big-endian, Q = unsigned 64-bit int
def enviar_mensaje(sock: socket.socket, texto: str):
cuerpo = texto.encode('utf-8')
cabecera = CABECERA.pack(len(cuerpo))
sock.sendall(cabecera + cuerpo)
def recibir_mensaje(sock: socket.socket) -> str:
cabecera_raw = _recibir_exacto(sock, CABECERA.size)
longitud = CABECERA.unpack(cabecera_raw)[0]
cuerpo = _recibir_exacto(sock, longitud)
return cuerpo.decode('utf-8')
def _recibir_exacto(sock: socket.socket, n: int) -> bytes:
"""Lee exactamente n bytes, esperando si llegan en trozos."""
datos = bytearray()
while len(datos) < n:
trozo = sock.recv(n - len(datos))
if not trozo:
raise ConnectionError("Conexión cerrada inesperadamente")
datos.extend(trozo)
return bytes(datos)
Timeouts y opciones de socket
import socket
with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
s.settimeout(5.0) # socket.timeout si la operación supera 5s
try:
s.connect(('example.com', 80))
s.sendall(b"GET / HTTP/1.0rnHost: example.comrnrn")
respuesta = b""
while True:
trozo = s.recv(4096)
if not trozo:
break
respuesta += trozo
print(respuesta[:200].decode('utf-8', errors='replace'))
except socket.timeout:
print("La conexión tardó demasiado")
except ConnectionRefusedError:
print("Conexión rechazada")
IPv6 y sockets Unix
import socket
# IPv6
with socket.socket(socket.AF_INET6, socket.SOCK_STREAM) as s:
s.connect(('::1', 9000))
# Socket Unix (solo Linux/macOS): comunicación entre procesos sin red
with socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) as s:
s.connect('/tmp/mi_servicio.sock')
Para la mayoría de los proyectos modernos, es preferible usar asyncio con sus primitivas de streams o socketserver de la biblioteca estándar. El módulo socket brilla cuando necesitas control total sobre el protocolo o implementas algo que ninguna biblioteca de alto nivel cubre.