Networking and Sockets in Python: TCP, UDP and HTTP

Networking and Sockets in Python: TCP, UDP and HTTP

Python provides low-level network access through the socket module and high-level abstractions via urllib, http.server, and third-party libraries like httpx. This guide covers all levels.

TCP sockets — client and server

import socket

HOST = '127.0.0.1'
PORT = 65432

# ── SERVER ────────────────────────────────────────────────────────────────────
def tcp_server():
    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"Server listening on {HOST}:{PORT}")

        while True:
            conn, addr = srv.accept()
            with conn:
                print(f"Connected by {addr}")
                while True:
                    data = conn.recv(4096)
                    if not data:
                        break
                    conn.sendall(data.upper())   # echo in uppercase

# ── CLIENT ────────────────────────────────────────────────────────────────────
def tcp_client(message: str) -> str:
    with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
        s.settimeout(10)
        s.connect((HOST, PORT))
        s.sendall(message.encode())
        response = s.recv(4096)
    return response.decode()

Multi-threaded TCP server

import socket
import threading

def handle_client(conn: socket.socket, addr: tuple):
    print(f"[+] New client: {addr}")
    with conn:
        while True:
            try:
                data = conn.recv(4096)
                if not data:
                    break
                conn.sendall(data.upper())
            except ConnectionResetError:
                break
    print(f"[-] Disconnected: {addr}")

def multithreaded_server(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"Multi-threaded server on {host}:{port}")
        while True:
            conn, addr = srv.accept()
            t = threading.Thread(target=handle_client, args=(conn, addr), daemon=True)
            t.start()

UDP sockets

import socket

def udp_server(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 listening on {host}:{port}")
        while True:
            data, addr = s.recvfrom(65536)
            print(f"UDP from {addr}: {data.decode()!r}")
            s.sendto(data.upper(), addr)

def udp_client(message: 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(message.encode(), (host, port))
        response, _ = s.recvfrom(65536)
    return response.decode()

Useful socket utilities

import socket

# DNS resolution
ip = socket.gethostbyname('example.com')
print(f"IP: {ip}")

# Full address info (IPv4 + IPv6)
infos = socket.getaddrinfo('example.com', 443, proto=socket.IPPROTO_TCP)
for info in infos:
    print(f"  {info[4][0]}:{info[4][1]} ({info[0].name})")

# Check if a port is open
def port_open(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"Port 80:  {port_open('example.com', 80)}")
print(f"Port 443: {port_open('example.com', 443)}")

# Get local IP
def local_ip() -> 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"Local IP: {local_ip()}")

Simple HTTP server

from http.server import HTTPServer, BaseHTTPRequestHandler
import json
import urllib.parse

class MyHandler(BaseHTTPRequestHandler):
    def log_message(self, fmt, *args):
        pass  # silence access logs

    def respond(self, code: int, body: str, ct: str = 'application/json'):
        data = body.encode('utf-8')
        self.send_response(code)
        self.send_header('Content-Type', f'{ct}; charset=utf-8')
        self.send_header('Content-Length', str(len(data)))
        self.end_headers()
        self.wfile.write(data)

    def do_GET(self):
        parsed = urllib.parse.urlparse(self.path)
        if parsed.path == '/':
            self.respond(200, json.dumps({'status': 'ok', 'version': '1.0'}))
        elif parsed.path == '/health':
            self.respond(200, json.dumps({'status': 'healthy'}))
        elif parsed.path.startswith('/echo/'):
            text = urllib.parse.unquote(parsed.path.removeprefix('/echo/'))
            self.respond(200, json.dumps({'echo': text}))
        else:
            self.respond(404, json.dumps({'error': 'not found'}))

    def do_POST(self):
        length = int(self.headers.get('Content-Length', 0))
        body   = self.rfile.read(length)
        try:
            data = json.loads(body)
            self.respond(200, json.dumps({'received': data, 'ok': True}))
        except json.JSONDecodeError:
            self.respond(400, json.dumps({'error': 'invalid JSON'}))

def start_server(host: str = '0.0.0.0', port: int = 8000):
    with HTTPServer((host, port), MyHandler) as httpd:
        print(f"HTTP server at http://{host}:{port}")
        httpd.serve_forever()

Modern HTTP client with httpx

pip install httpx

import httpx
import asyncio

# Synchronous client with retries and granular timeouts
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': 'MyApp/1.0'},
    ) as client:
        resp = client.get(url, params=params)
        resp.raise_for_status()
        return resp.json()

data = fetch_api('https://jsonplaceholder.typicode.com/posts', params={'_limit': 5})
print(f"Received: {len(data)} posts")

# Async client — concurrent requests
async def fetch_many(urls: list[str]) -> list[dict]:
    async with httpx.AsyncClient(timeout=30) as client:
        responses = await asyncio.gather(*[
            client.get(url) for url in urls
        ], return_exceptions=True)

    results = []
    for url, resp in zip(urls, responses):
        if isinstance(resp, Exception):
            print(f"Error {url}: {resp}")
        elif resp.status_code == 200:
            results.append({'url': url, 'data': resp.json()})
    return results

urls    = [f"https://jsonplaceholder.typicode.com/posts/{i}" for i in range(1, 11)]
results = asyncio.run(fetch_many(urls))
print(f"Downloaded: {len(results)} posts")

WebSockets

pip install websockets

import asyncio
import websockets
import json

# ── SERVER ────────────────────────────────────────────────────────────────────
connected = set()

async def handle_ws(websocket):
    connected.add(websocket)
    print(f"Client connected: {websocket.remote_address}")
    try:
        async for message in websocket:
            data = json.loads(message)
            broadcast = json.dumps({
                'type': 'message',
                'from': str(websocket.remote_address),
                'text': data.get('text', ''),
            })
            others = connected - {websocket}
            if others:
                await asyncio.gather(*[ws.send(broadcast) for ws in others])
    except websockets.exceptions.ConnectionClosed:
        pass
    finally:
        connected.discard(websocket)

async def ws_server():
    async with websockets.serve(handle_ws, '0.0.0.0', 8765):
        print("WebSocket server on ws://0.0.0.0:8765")
        await asyncio.Future()

# ── CLIENT ────────────────────────────────────────────────────────────────────
async def ws_client(uri: str, messages: list[str]):
    async with websockets.connect(uri) as ws:
        for msg in messages:
            await ws.send(json.dumps({'text': msg}))
            try:
                reply = await asyncio.wait_for(ws.recv(), timeout=1.0)
                print(f"Reply: {reply}")
            except asyncio.TimeoutError:
                pass

urllib — zero external dependencies

import urllib.request
import json

with urllib.request.urlopen('https://httpbin.org/get', timeout=10) as r:
    data = json.loads(r.read())
    print(f"Origin IP: {data['origin']}")

# POST with JSON
payload = json.dumps({'key': 'value'}).encode()
req = urllib.request.Request(
    'https://httpbin.org/post',
    data=payload,
    headers={'Content-Type': 'application/json'},
)
with urllib.request.urlopen(req, timeout=10) as r:
    result = json.loads(r.read())
    print(f"JSON back: {result['json']}")

# Download with progress
def download(url: str, dest: str):
    def report(blocks, block_size, total):
        if total > 0:
            pct = blocks * block_size / total * 100
            print(f"\r  {min(pct,100):.1f}%", end='', flush=True)
    urllib.request.urlretrieve(url, dest, reporthook=report)
    print()

Best practices

• socket.setsockopt(SOL_SOCKET, SO_REUSEADDR, 1) on servers to avoid "address already in use" after a quick restart.
• s.settimeout(n) on all client sockets — a socket without a timeout can block forever.
• httpx over requests for new code: supports HTTP/2, native async, and granular per-phase timeouts.
• WebSockets for real-time bidirectional communication (chat, notifications, live dashboards); HTTP for classic request-response.
• Always limit incoming data size — never read unbounded from a socket to prevent memory-exhaustion attacks.
• socket.AF_INET6 with SOCK_STREAM for IPv6 support; '0.0.0.0' binds IPv4 only, '::' accepts both on dual-stack systems.

Related conversions

Frequent conversions across the catalogue:

• JPG to PNG
• MP4 to MP3
• PDF to DOCX
• PNG to JPG
• MP3 to WAV
• DOCX to PDF