Redes y Sockets en Python: TCP, UDP y HTTP
Python proporciona acceso de bajo nivel a la red mediante el módulo socket y abstracciones de alto nivel con urllib, http.server y bibliotecas de terceros como httpx. Esta guía recorre todos los niveles.
Sockets TCP — cliente y servidor
import socket
HOST = '127.0.0.1'
PORT = 65432
# ── SERVIDOR ──────────────────────────────────────────────────────────────────
def servidor_tcp():
with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as srv:
srv.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
srv.bind((HOST, PORT))
srv.listen(5)
print(f"Servidor escuchando en {HOST}:{PORT}")
while True:
conn, addr = srv.accept()
with conn:
print(f"Conexión de {addr}")
while True:
datos = conn.recv(4096)
if not datos:
break
print(f"Recibido: {datos.decode()!r}")
conn.sendall(datos.upper()) # eco en mayúsculas
# ── CLIENTE ───────────────────────────────────────────────────────────────────
def cliente_tcp(mensaje: str) -> str:
with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
s.settimeout(10)
s.connect((HOST, PORT))
s.sendall(mensaje.encode())
respuesta = s.recv(4096)
return respuesta.decode()
# respuesta = cliente_tcp("hola mundo")
# print(respuesta) # "HOLA MUNDO"
Servidor TCP multihilo
import socket
import threading
def manejar_cliente(conn: socket.socket, addr: tuple):
"""Hilo dedicado por cliente — permite múltiples conexiones simultáneas."""
print(f"[+] Nuevo cliente: {addr}")
with conn:
while True:
try:
datos = conn.recv(4096)
if not datos:
break
conn.sendall(datos.upper())
except ConnectionResetError:
break
print(f"[-] Desconectado: {addr}")
def servidor_multihilo(host: str = '0.0.0.0', port: int = 65432):
with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as srv:
srv.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
srv.bind((host, port))
srv.listen(50)
print(f"Servidor multihilo en {host}:{port}")
while True:
conn, addr = srv.accept()
hilo = threading.Thread(target=manejar_cliente, args=(conn, addr), daemon=True)
hilo.start()
# servidor_multihilo()
Sockets UDP
import socket
# ── SERVIDOR UDP ──────────────────────────────────────────────────────────────
def servidor_udp(host: str = '0.0.0.0', port: int = 65433):
with socket.socket(socket.AF_INET, socket.SOCK_DGRAM) as s:
s.bind((host, port))
print(f"UDP escuchando en {host}:{port}")
while True:
datos, addr = s.recvfrom(65536)
print(f"UDP de {addr}: {datos.decode()!r}")
s.sendto(datos.upper(), addr) # respuesta al mismo cliente
# ── CLIENTE UDP ───────────────────────────────────────────────────────────────
def cliente_udp(mensaje: str, host: str = '127.0.0.1', port: int = 65433) -> str:
with socket.socket(socket.AF_INET, socket.SOCK_DGRAM) as s:
s.settimeout(5)
s.sendto(mensaje.encode(), (host, port))
respuesta, _ = s.recvfrom(65536)
return respuesta.decode()
Operaciones de red de alto nivel — socket útil
import socket
# Resolver nombre de dominio
ip = socket.gethostbyname('kaijuconverter.com')
print(f"IP: {ip}")
# Información completa (para IPv4 e IPv6)
infos = socket.getaddrinfo('kaijuconverter.com', 443, proto=socket.IPPROTO_TCP)
for info in infos:
print(f" {info[4][0]}:{info[4][1]} ({info[0].name})")
# Comprobar si un puerto está abierto
def puerto_abierto(host: str, port: int, timeout: float = 2.0) -> bool:
with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
s.settimeout(timeout)
return s.connect_ex((host, port)) == 0
print(f"Puerto 80 abierto: {puerto_abierto('kaijuconverter.com', 80)}")
print(f"Puerto 443 abierto: {puerto_abierto('kaijuconverter.com', 443)}")
# Obtener IP local
def obtener_ip_local() -> str:
with socket.socket(socket.AF_INET, socket.SOCK_DGRAM) as s:
s.connect(('8.8.8.8', 80))
return s.getsockname()[0]
print(f"IP local: {obtener_ip_local()}")
Servidor HTTP simple con http.server
from http.server import HTTPServer, BaseHTTPRequestHandler
import json
import urllib.parse
class MiHandler(BaseHTTPRequestHandler):
# Silenciar logs de acceso (opcional)
def log_message(self, format, *args):
pass
def responder(self, codigo: int, cuerpo: str, content_type: str = 'application/json'):
data = cuerpo.encode('utf-8')
self.send_response(codigo)
self.send_header('Content-Type', f'{content_type}; charset=utf-8')
self.send_header('Content-Length', str(len(data)))
self.end_headers()
self.wfile.write(data)
def do_GET(self):
ruta = urllib.parse.urlparse(self.path)
params = urllib.parse.parse_qs(ruta.query)
if ruta.path == '/':
self.responder(200, json.dumps({'estado': 'ok', 'version': '1.0'}))
elif ruta.path == '/health':
self.responder(200, json.dumps({'status': 'healthy'}))
elif ruta.path.startswith('/eco/'):
texto = ruta.path.removeprefix('/eco/')
self.responder(200, json.dumps({'eco': urllib.parse.unquote(texto)}))
else:
self.responder(404, json.dumps({'error': 'ruta no encontrada'}))
def do_POST(self):
longitud = int(self.headers.get('Content-Length', 0))
cuerpo = self.rfile.read(longitud)
try:
datos = json.loads(cuerpo)
self.responder(200, json.dumps({'recibido': datos, 'ok': True}))
except json.JSONDecodeError:
self.responder(400, json.dumps({'error': 'JSON inválido'}))
def iniciar_servidor(host: str = '0.0.0.0', port: int = 8000):
with HTTPServer((host, port), MiHandler) as httpd:
print(f"HTTP servidor en http://{host}:{port}")
httpd.serve_forever()
# iniciar_servidor()
Cliente HTTP moderno con httpx
pip install httpx
import httpx
import asyncio
# Cliente síncrono con reintentos automáticos y timeout
def fetch_api(url: str, params: dict | None = None) -> dict:
transport = httpx.HTTPTransport(retries=3)
with httpx.Client(
transport=transport,
timeout=httpx.Timeout(connect=5, read=30, write=10, pool=5),
headers={'User-Agent': 'MiApp/1.0'},
) as client:
resp = client.get(url, params=params)
resp.raise_for_status()
return resp.json()
datos = fetch_api('https://jsonplaceholder.typicode.com/posts', params={'_limit': 5})
print(f"Recibidos: {len(datos)} posts")
# Cliente asíncrono — peticiones concurrentes
async def fetch_muchos(urls: list[str]) -> list[dict]:
async with httpx.AsyncClient(timeout=30) as client:
respuestas = await asyncio.gather(*[
client.get(url) for url in urls
], return_exceptions=True)
resultados = []
for url, resp in zip(urls, respuestas):
if isinstance(resp, Exception):
print(f"Error {url}: {resp}")
elif resp.status_code == 200:
resultados.append({'url': url, 'datos': resp.json()})
return resultados
urls = [f"https://jsonplaceholder.typicode.com/posts/{i}" for i in range(1, 11)]
resultados = asyncio.run(fetch_muchos(urls))
print(f"Descargados: {len(resultados)} posts")
WebSockets con websockets
pip install websockets
import asyncio
import websockets
import json
# ── SERVIDOR WebSocket ────────────────────────────────────────────────────────
clientes_conectados = set()
async def manejar_ws(websocket):
"""Servidor de chat simple con broadcast."""
clientes_conectados.add(websocket)
print(f"Cliente conectado: {websocket.remote_address}")
try:
async for mensaje in websocket:
datos = json.loads(mensaje)
respuesta = json.dumps({
'tipo': 'mensaje',
'de': str(websocket.remote_address),
'texto': datos.get('texto', ''),
})
# Broadcast a todos los clientes excepto el emisor
destinatarios = clientes_conectados - {websocket}
if destinatarios:
await asyncio.gather(*[ws.send(respuesta) for ws in destinatarios])
except websockets.exceptions.ConnectionClosed:
pass
finally:
clientes_conectados.discard(websocket)
print(f"Cliente desconectado: {websocket.remote_address}")
async def servidor_ws():
async with websockets.serve(manejar_ws, '0.0.0.0', 8765):
print("WebSocket servidor en ws://0.0.0.0:8765")
await asyncio.Future() # esperar indefinidamente
# ── CLIENTE WebSocket ─────────────────────────────────────────────────────────
async def cliente_ws(uri: str, mensajes: list[str]):
async with websockets.connect(uri) as ws:
for msg in mensajes:
await ws.send(json.dumps({'texto': msg}))
print(f"Enviado: {msg!r}")
try:
respuesta = await asyncio.wait_for(ws.recv(), timeout=1.0)
print(f"Recibido: {respuesta}")
except asyncio.TimeoutError:
pass # sin respuesta en 1s
# asyncio.run(servidor_ws())
# asyncio.run(cliente_ws('ws://localhost:8765', ['Hola', 'Mundo']))
urllib — sin dependencias externas
import urllib.request
import urllib.parse
import json
# GET simple
with urllib.request.urlopen('https://httpbin.org/get', timeout=10) as resp:
datos = json.loads(resp.read())
print(f"IP origen: {datos['origin']}")
# POST con JSON
payload = json.dumps({'clave': 'valor'}).encode('utf-8')
req = urllib.request.Request(
'https://httpbin.org/post',
data=payload,
headers={'Content-Type': 'application/json'},
)
with urllib.request.urlopen(req, timeout=10) as resp:
resultado = json.loads(resp.read())
print(f"Recibido JSON: {resultado['json']}")
# Descargar archivo con progreso
def descargar_con_progreso(url: str, destino: str):
def reporte(bloques, tam_bloque, tam_total):
if tam_total > 0:
pct = bloques * tam_bloque / tam_total * 100
print(f"\r {min(pct, 100):.1f}%", end='', flush=True)
urllib.request.urlretrieve(url, destino, reporthook=reporte)
print() # salto de línea tras el progreso
Buenas prácticas
- Usar
socket.setsockopt(SOL_SOCKET, SO_REUSEADDR, 1)en servidores para evitar el error "address already in use" al reiniciar rápidamente. s.settimeout(n)en todos los sockets cliente — un socket sin timeout puede bloquear indefinidamente.- httpx sobre requests para código nuevo: soporta HTTP/2, async nativo y timeouts granulares por fase (connect, read, write).
- WebSockets para comunicación bidireccional en tiempo real (chat, notificaciones, dashboards); HTTP para petición-respuesta clásica.
- Nunca confiar en
Content-Lengthsin validar al recibir datos por red — limitar el tamaño máximo de lectura para prevenir ataques de agotamiento de memoria. socket.AF_INET6consocket.SOCK_STREAMpara soportar IPv6; usar'0.0.0.0'limita a IPv4,'::'acepta ambas versiones en sistemas con dual-stack.
Conversiones relacionadas
Conversiones frecuentes del catálogo: