Built-In HTTP Networking: Servers, Clients, and Request Handlers

Updated

September 7, 2026

Built-In HTTP Networking: Servers, Clients, and Request Handlers

After reading this chapter, you will master Python’s zero-dependency HTTP networking stack: spin up lightweight microservices using http.server.ThreadingHTTPServer, parse JSON payloads and query parameters in BaseHTTPRequestHandler, execute performant client requests using http.client and urllib.request, and prevent concurrency starvation bugs.

Mental model

While third-party packages like requests, httpx, and FastAPI are common in large applications, Python’s built-in standard library provides a complete, production-grade HTTP implementation requiring zero external dependencies:

CPython Built-in HTTP Stack Architecture:
┌────────────────────────────────────────────────────────┐
│ High-Level Abstractions                                │
│   Server: http.server.ThreadingHTTPServer              │
│   Client: urllib.request.urlopen                       │
└────────────────────────┬───────────────────────────────┘
                         │
                         ▼
┌────────────────────────────────────────────────────────┐
│ Protocol Framing Layer                                 │
│   Server: http.server.BaseHTTPRequestHandler           │
│   Client: http.client.HTTPConnection / HTTPSConnection │
└────────────────────────┬───────────────────────────────┘
                         │
                         ▼
┌────────────────────────────────────────────────────────┐
│ Low-Level Transport Layer                              │
│   socketserver.TCPServer ──▶ socket.socket             │
└────────────────────────────────────────────────────────┘

Minimal example

Save as http_service_demo.py:

# http_service_demo.py
import json
import threading
import urllib.request
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer

class HealthCheckHandler(BaseHTTPRequestHandler):
    """Zero-dependency HTTP handler serving JSON health endpoints."""

    def do_GET(self) -> None:
        if self.path == "/health":
            payload = json.dumps({"status": "SERVING", "version": "1.0.0"}).encode("utf-8")
            self.send_response(200)
            self.send_header("Content-Type", "application/json")
            self.send_header("Content-Length", str(len(payload)))
            self.end_headers()
            self.wfile.write(payload)
        else:
            self.send_error(404, "Endpoint Not Found")

    def log_message(self, format: str, *args: object) -> None:
        # Suppress noisy default standard error logging
        pass

def main() -> None:
    # 1. Bind to ephemeral local port (port 0 selects available random port)
    server = ThreadingHTTPServer(("127.0.0.1", 0), HealthCheckHandler)
    host, port = server.server_address

    # 2. Run server loop on a background thread
    server_thread = threading.Thread(target=server.serve_forever, daemon=True)
    server_thread.start()
    print(f"HTTP Server active on http://{host}:{port}")

    # 3. Query the endpoint using built-in urllib.request
    try:
        url = f"http://{host}:{port}/health"
        with urllib.request.urlopen(url, timeout=3) as response:
            status_code = response.status
            body = json.loads(response.read().decode("utf-8"))

        print(f"Client received HTTP {status_code}: {body}")
    finally:
        server.shutdown()
        server.server_close()
        print("HTTP Server cleanly stopped.")

if __name__ == "__main__":
    main()

Run via uv run python http_service_demo.py:

HTTP Server active on http://127.0.0.1:41829
Client received HTTP 200: {'status': 'SERVING', 'version': '1.0.0'}
HTTP Server cleanly stopped.

Worked examples

Case 1: Building a REST Metrics Ingestion Webhook Handler

Production microservices frequently expose webhooks to accept JSON telemetry records via POST requests while reporting metrics on GET:

# metrics_webhook_service.py
import json
import threading
import urllib.request
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer

class MetricsCollectorHandler(BaseHTTPRequestHandler):
    # Shared in-memory metrics storage
    metrics_store: dict[str, float] = {}
    lock = threading.Lock()

    def do_POST(self) -> None:
        if self.path == "/api/v1/metrics":
            # Read Content-Length to determine exact body bytes to read
            content_length = int(self.headers.get("Content-Length", 0))
            if content_length == 0:
                self.send_error(400, "Missing or zero Content-Length")
                return

            body = self.rfile.read(content_length)
            try:
                data = json.loads(body.decode("utf-8"))
            except json.JSONDecodeError:
                self.send_error(400, "Invalid JSON payload")
                return

            # Thread-safe write
            with self.lock:
                self.metrics_store.update(data)

            ack_payload = json.dumps({"acknowledged_keys": list(data.keys())}).encode("utf-8")
            self.send_response(202)
            self.send_header("Content-Type", "application/json")
            self.send_header("Content-Length", str(len(ack_payload)))
            self.end_headers()
            self.wfile.write(ack_payload)
        else:
            self.send_error(404)

    def log_message(self, format: str, *args: object) -> None:
        pass

def main() -> None:
    server = ThreadingHTTPServer(("127.0.0.1", 0), MetricsCollectorHandler)
    host, port = server.server_address
    t = threading.Thread(target=server.serve_forever, daemon=True)
    t.start()

    try:
        # Client POST request using standard library
        url = f"http://{host}:{port}/api/v1/metrics"
        sample_payload = json.dumps({"cpu_util": 84.2, "mem_util": 61.0}).encode("utf-8")
        req = urllib.request.Request(
            url,
            data=sample_payload,
            headers={"Content-Type": "application/json"},
            method="POST",
        )

        with urllib.request.urlopen(req, timeout=3) as resp:
            print(f"POST Status: {resp.status}")
            print(f"Response:    {resp.read().decode('utf-8')}")

        print(f"\nServer-side metrics store: {MetricsCollectorHandler.metrics_store}")
    finally:
        server.shutdown()
        server.server_close()

if __name__ == "__main__":
    main()

Run:

uv run python metrics_webhook_service.py

Output:

POST Status: 202
Response:    {"acknowledged_keys": ["cpu_util", "mem_util"]}

Server-side metrics store: {'cpu_util': 84.2, 'mem_util': 61.0}

Case 2: Low-Level Streaming Requests with http.client.HTTPConnection

When handling large file downloads or high-throughput API communication, http.client provides direct, low-level access to the socket stream without loading the entire payload into RAM:

# http_client_streaming.py
import http.client
import json
import threading
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer

class BulkDataHandler(BaseHTTPRequestHandler):
    def do_GET(self) -> None:
        if self.path == "/stream":
            self.send_response(200)
            self.send_header("Content-Type", "text/plain")
            self.end_headers()
            # Stream 3 chunks sequentially
            for i in range(1, 4):
                self.wfile.write(f"Chunk-#{i}\n".encode("utf-8"))
        else:
            self.send_error(404)

    def log_message(self, *args: object) -> None:
        pass

def main() -> None:
    server = ThreadingHTTPServer(("127.0.0.1", 0), BulkDataHandler)
    host, port = server.server_address
    threading.Thread(target=server.serve_forever, daemon=True).start()

    try:
        # Use http.client for granular socket streaming
        conn = http.client.HTTPConnection(host, port, timeout=5)
        conn.request("GET", "/stream")
        response = conn.getresponse()

        print(f"Response Status: {response.status} {response.reason}")
        print("Reading streamed chunks on client:")
        while chunk := response.readline():
            print(f"  Received: {chunk.decode('utf-8').strip()}")
        conn.close()
    finally:
        server.shutdown()
        server.server_close()

if __name__ == "__main__":
    main()

Run:

uv run python http_client_streaming.py

Output:

Response Status: 200 OK
Reading streamed chunks on client:
  Received: Chunk-#1
  Received: Chunk-#2
  Received: Chunk-#3

Pitfalls

Pitfall 1: Using HTTPServer Instead of ThreadingHTTPServer

The legacy http.server.HTTPServer is strictly single-threaded. If a single client connection hangs or takes 10 seconds to upload, all other clients are blocked entirely:

# THE TRAP: Single-threaded; one slow client starves the entire server!
from http.server import HTTPServer, BaseHTTPRequestHandler
# server = HTTPServer(('0.0.0.0', 8080), MyHandler)

# THE FIX: Always use ThreadingHTTPServer (handles each request in a worker thread)
from http.server import ThreadingHTTPServer
server = ThreadingHTTPServer(('0.0.0.0', 8080), BaseHTTPRequestHandler)

Pitfall 2: Omitting the timeout Parameter in urllib.request.urlopen

By default, urllib.request.urlopen uses the global socket default timeout (None), meaning a client connection can hang forever if the remote host drops the TCP packet:

# THE TRAP:
# with urllib.request.urlopen("https://example.com") as resp: ...  # Can hang indefinitely!

# THE FIX: Always specify an explicit timeout in seconds
import urllib.request
with urllib.request.urlopen("https://example.com", timeout=5.0) as resp:
    data = resp.read()

Pitfall 3: Not Reading Content-Length in do_POST

In do_POST, calling self.rfile.read() without specifying the number of bytes to read will hang indefinitely because the socket connection remains open for keep-alive:

# THE TRAP:
def do_POST(self):
    # data = self.rfile.read()  # HANGS! Socket is waiting for EOF that never arrives!

# THE FIX: Read only up to Content-Length bytes:
def do_POST(self):
    length = int(self.headers.get('Content-Length', 0))
    data = self.rfile.read(length)

Exercises

  1. Build a zero-dependency HTTP server that serves static files from a local directory, returning 404 Not Found if the file does not exist.
  2. Implement an HTTP handler that parses query parameters (e.g. /search?q=python&limit=10) using urllib.parse.parse_qs.
  3. Write a script that uses http.client.HTTPConnection to send a HEAD request to inspect response headers without downloading the body.
  4. Implement basic HTTP authentication verification (Authorization: Basic ...) in a custom BaseHTTPRequestHandler.
  5. Measure the throughput in requests-per-second of ThreadingHTTPServer when queried by 10 concurrent threads using urllib.request.

Further reading

  • Python Documentation: http.server — HTTP servers.
  • Python Documentation: http.client — HTTP protocol client.
  • Python Documentation: urllib.request — Extensible library for opening URLs.
  • RFC 9110: HTTP Semantics.