uHTTP Server

Micro HTTP server for MicroPython and CPython.

Features

• MicroPython and CPython compatible
• Low-level POSIX socket implementation
• Fully synchronous but handles multiple connections
• Delayed response support (hold client, reply later)
• Raw data (HTML, binary) and JSON support
• SSL/TLS for HTTPS connections
• IPv6 and dual-stack support
• Event mode for streaming large uploads
• WebSocket support (RFC 6455)
• Server-Sent Events (SSE) streaming
• NDJSON streaming responses (application/x-ndjson)
• Memory-efficient (~32KB RAM minimum)

Installation

pip install uhttp-server

For MicroPython, copy uhttp/server.py to your device.

Usage

import uhttp.server

server = uhttp.server.HttpServer(port=9980)

while True:
    client = server.wait()
    if client:
        if client.path == '/':
            # result is html
            client.respond("<h1>hello</h1><p>uHTTP</p>")
        elif client.path == '/rpc':
            # result is json
            client.respond({'message': 'hello', 'success': True, 'headers': client.headers, 'query': client.query})
        else:
            client.respond("Not found", status=404)

Migrating from 2.x to 3.0

3.0 replaces the internal select.select() loop with a selectors selector that the server, its connections and its WebSockets register themselves in. Dispatch goes through the owner stored in key.data, so one loop can drive several servers, their connections and other subsystems (including uhttp-client 3.x) together.

Removed: read_sockets, write_sockets, process_events(), event_read(), event_write() — on HttpServer and on WebSocket.

server.wait(timeout) is unchanged, so single-server code needs no edits.

External loop

# 2.x
while True:
    r, w, _ = select.select(
        http.read_sockets + https.read_sockets,
        http.write_sockets + https.write_sockets, [], 1.0)
    client = http.process_events(r, w) or https.process_events(r, w)
    if client:
        serve(client)

# 3.0
selector = selectors.DefaultSelector()
http = HttpServer(port=80, selector=selector)
https = HttpServer(port=443, ssl_context=ctx, selector=selector)

while True:
    for key, mask in selector.select(1.0):
        client = key.data.handle_event(key.fileobj, mask)
        if isinstance(client, HttpConnection):
            serve(client)
            while client.next():   # drain events buffered from one recv
                serve(client)
    http.maintenance()
    https.maintenance()

Three rules carry the whole migration:

1. key.data.handle_event(key.fileobj, mask) replaces process_events(). It returns the connection when a request/event is ready, otherwise None. Anything you register yourself must expose the same method — register an owner object, not a bare callback.
2. while client.next(): — one recv() can carry several WebSocket frames or body chunks, and select() will not report them again because the OS buffer is already drained. Drain them before blocking. wait() does this for you.
3. maintenance() once per loop iteration — keep-alive and header timeouts have no event to ride on. wait() calls it for you.

WebSocket

Non-event mode still returns a WebSocket with the same wait(), read_buffer(), send() and is_closed, but it is now a facade over the connection — so it honours max_send_buffer_size and send()/ping() return False when they cannot queue. To drive it from a shared loop, hand the selector to the upgrade:

ws = client.accept_websocket(selector=selector)   # joins the shared loop
ws = client.accept_websocket()                    # owns one, use ws.wait()

wait() refuses a shared selector: a blocking wait cannot service the other owners' ready keys.

Not yet on MicroPython

selectors is not in the MicroPython standard library, so 3.x is CPython-only until a shim lands. Stay on 2.x on-device.

SSL/HTTPS Support

uHTTP supports SSL/TLS encryption for HTTPS connections on both CPython and MicroPython.

Basic HTTPS Server

import ssl
import uhttp.server

# Create SSL context
context = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER)
context.load_cert_chain(certfile='cert.pem', keyfile='key.pem')

# Create HTTPS server
server = uhttp.server.HttpServer(port=443, ssl_context=context)

while True:
    client = server.wait()
    if client:
        # Check if connection is secure
        if client.is_secure:
            client.respond({'message': 'Secure HTTPS connection!'})
        else:
            client.respond({'message': 'Insecure HTTP connection'})

Using Let's Encrypt / Certbot Certificates

Certbot creates certificates in /etc/letsencrypt/live/your-domain/ with these files:

• cert.pem - Your domain certificate only
• chain.pem - Certificate authority chain
• fullchain.pem - Your certificate + CA chain (use this for certfile)
• privkey.pem - Private key (use this for keyfile)

Important: Always use fullchain.pem (not cert.pem) as the certificate file. Without the full chain, clients will get "certificate verification failed" errors.

Example with Certbot Certificates

import ssl
import uhttp.server

# Create SSL context with Let's Encrypt certificates
context = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER)
context.load_cert_chain(
    certfile='/etc/letsencrypt/live/example.com/fullchain.pem',
    keyfile='/etc/letsencrypt/live/example.com/privkey.pem'
)

# Create HTTPS server
server = uhttp.server.HttpServer(
    address='0.0.0.0',
    port=443,
    ssl_context=context
)

while True:
    client = server.wait()
    if client:
        client.respond({'message': 'Hello from HTTPS!'})

Permissions Note

The /etc/letsencrypt/ directory requires root access. You have two options:

1. Run as root (not recommended for production):

   sudo python3 your_server.py
   

2. Copy certificates to accessible location (recommended):

   # Copy certificates to your application directory
   sudo cp /etc/letsencrypt/live/example.com/fullchain.pem ~/myapp/
   sudo cp /etc/letsencrypt/live/example.com/privkey.pem ~/myapp/
   sudo chown youruser:youruser ~/myapp/*.pem
   sudo chmod 600 ~/myapp/privkey.pem
   

   Then use the copied files:

   context.load_cert_chain(
       certfile='/home/youruser/myapp/fullchain.pem',
       keyfile='/home/youruser/myapp/privkey.pem'
   )
   

Certificate Renewal

Let's Encrypt certificates expire every 90 days. After renewal with certbot renew, restart your server to load the new certificates, or implement a reload mechanism:

# Renew certificates
sudo certbot renew

# Restart your application
sudo systemctl restart your-app

HTTP to HTTPS Redirect

Run both HTTP and HTTPS servers to redirect HTTP traffic:

import ssl
import selectors
import uhttp.server

# SSL context for HTTPS
context = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER)
context.load_cert_chain(
    certfile='/etc/letsencrypt/live/example.com/fullchain.pem',
    keyfile='/etc/letsencrypt/live/example.com/privkey.pem'
)

# One shared selector drives both servers. On a ready key,
# key.data.handle_event() dispatches to the owning server/connection.
selector = selectors.DefaultSelector()

# HTTP server (redirects)
http_server = uhttp.server.HttpServer(port=80, selector=selector)

# HTTPS server (serves content)
https_server = uhttp.server.HttpServer(
    port=443, ssl_context=context, selector=selector)

def serve(client):
    if client.is_secure:
        client.respond({'message': 'Secure content'})   # HTTPS content
    else:
        client.respond_redirect(f"https://{client.host}{client.url}")  # redirect

while True:
    for key, mask in selector.select(1.0):
        client = key.data.handle_event(key.fileobj, mask)
        if isinstance(client, uhttp.server.HttpConnection):
            serve(client)
            while client.next():   # drain events buffered from one recv (WS/stream)
                serve(client)
    http_server.maintenance()
    https_server.maintenance()

Testing SSL Locally

For local development, create self-signed certificates:

openssl req -x509 -newkey rsa:2048 -keyout key.pem -out cert.pem -days 365 -nodes -subj "/CN=localhost"

Then use them:

context.load_cert_chain(certfile='cert.pem', keyfile='key.pem')

Test with curl (use -k to accept self-signed certificates):

curl -k https://localhost:8443/

See examples/ directory for complete working examples.

API

General methods:

import uhttp.server

uhttp.server.decode_percent_encoding(data)

• Decode percent encoded data (bytes)

uhttp.server.parse_header_parameters(value)

• Parse parameters/directives from header value, returns dict

uhttp.server.parse_cookies(raw_cookies)

• Parse cookie string into dict

uhttp.server.parse_query(raw_query, query=None)

• Parse raw_query from URL, append it to existing query, returns dict

uhttp.server.parse_url(url)

• Parse URL to path and query

uhttp.server.parse_header_line(line)

• Parse header line to key and value

uhttp.server.encode_response_data(headers, data)

• Encode response data by its type

Class HttpServer:

HttpServer(address='0.0.0.0', port=80, ssl_context=None, event_mode=False, **kwargs)

Parameters:

• address - IP address to bind to (default: '0.0.0.0')
• port - Port to listen on (default: 80)
• ssl_context - Optional ssl.SSLContext for HTTPS connections (default: None)
• event_mode - Enable event mode for streaming uploads (default: False)
• **kwargs - Additional options:
  • max_waiting_clients - Maximum concurrent connections (default: 32)
  • keep_alive_timeout - Keep-alive timeout in seconds (default: 15)
  • keep_alive_max_requests - Max requests per connection (default: 100)
  • request_timeout - Max seconds from first byte to complete headers (default: 5)
  • max_headers_length - Maximum header size in bytes (default: 4KB)
  • max_content_length - Maximum body size in bytes (default: 512KB, only enforced when event_mode=False)
  • max_send_buffer_size - Maximum pending bytes in send buffer for backpressure (default: 64KB). When a slow client cannot drain TCP fast enough, _send() raises OSError instead of growing _send_buffer unbounded.
  • max_ws_message_length - Maximum WebSocket message size before chunking (default: 64KB)
  • file_chunk_size - Chunk size in bytes for streaming file responses (default: 4KB)
  • listen - Listening socket backlog (default: 8)
  • trusted_proxies - List of trusted proxy IP addresses (default: None). When set, remote_address uses X-Forwarded-For header for connections from these IPs. When not set, X-Forwarded-For is ignored.
  • selector - A selectors.BaseSelector to register sockets in (default: a DefaultSelector the server owns and closes). Pass the same instance to several servers (and register your own sockets in it) to drive them from one loop.

Properties:

socket(self)

• Server socket

selector(self)

• The selectors.BaseSelector the server registers its sockets in. Read events from it and dispatch via key.data.handle_event() for a shared-selector loop across several servers / your own sockets (register those with an owner object exposing handle_event(fileobj, mask)). Pass selector= to the constructor to share one; wait() on a shared selector services only this server's sockets.

is_secure(self)

• Returns True if server uses SSL/TLS, False otherwise

event_mode(self)

• Returns True if event mode is enabled

Methods:

wait(self, timeout=1)

• Wait for socket activity with the given timeout, returns None or an HttpConnection with an established request. Single-server convenience backed by the server's own selector; internally accepts connections, flushes writes, and runs maintenance().

handle_event(self, fileobj, mask)

• Owner dispatch for the server's listening socket (stored as key.data). Accepts a new connection and returns None. Used by a shared-selector loop. (Connections stored as key.data expose the same method; theirs returns the connection when a request is ready.)

maintenance(self)

• Closes idle / timed-out connections. wait() calls it after each active tick; call it once per loop iteration when driving a shared selector yourself.

Class HttpConnection:

HttpConnection(server, sock, addr, **kwargs)

Properties:

addr(self)

• Client address tuple (ip, port)

socket_address(self)

• Client socket address as string ip:port (always socket IP, ignores X-Forwarded-For)

remote_address(self)

• Client remote address. Returns first IP from X-Forwarded-For if connection is from trusted_proxies, otherwise socket address.

remote_addresses(self)

• Full X-Forwarded-For chain if from trusted proxy, otherwise socket address.

method(self)

• HTTP method

url(self)

• URL address

host(self)

• URL address

full_url(self)

• URL address

protocol(self)

• Protocol

headers(self)

• headers dict

data(self)

• Content data

path(self)

• Path

query(self)

• Query dict

cookies(self)

• Cookies dict

is_secure(self)

• Returns True if connection is using SSL/TLS, False otherwise

socket(self)

• This socket

is_loaded(self)

• Returns True when request is fully loaded and ready for response

content_length(self)

• Content length

event(self) (event mode only)

• Current event type: EVENT_REQUEST, EVENT_HEADERS, EVENT_DATA, EVENT_COMPLETE, or EVENT_ERROR

bytes_received(self) (event mode only)

• Number of body bytes received so far

error(self) (event mode only)

• Error message when event is EVENT_ERROR

context (event mode only)

• Application storage attribute for request state (read-write)

is_websocket_request(self)

• True if request is a WebSocket upgrade request

is_websocket(self)

• True if connection is in WebSocket mode

ws_message(self)

• Last received WebSocket message (str for text frames, bytes for binary frames)

Methods:

headers_get(self, key, default=None)

• Return value from headers by key, or default if key not found
• Repeated field lines are combined into one value per RFC 9110 5.3 — with , , except Cookie, which RFC 6265 separates with ;

headers_all(self, key)

• Return every value of a repeated header as a list, in the order received, or [] if the header was not sent. Use it when combining would be ambiguous, e.g. a value that may itself contain a comma:

client.headers_get('accept')   # 'text/html, application/xml, text/plain'
client.headers_all('accept')   # ['text/html, application/xml', 'text/plain']

process_request(self)

• Process HTTP request when read event on client socket

handle_event(self, fileobj, mask)

• Owner dispatch for this connection (stored as key.data). Drives read/write for the given readiness mask and returns the connection when a request/event is ready, else None. Used by a shared-selector loop.

next(self)

• Process the next event already in the receive buffer, returning True while another event is ready on this same connection. One recv() can pull several WebSocket frames / body chunks that select() will not report again; call next() in a loop after handling an event to drain them. Returns False on a plain HTTP connection or after a non-event WebSocket handoff.

respond(self, data=None, status=200, headers=None, cookies=None)

• Create general response with data, status and headers as dict

respond_redirect(self, url, status=302, cookies=None)

• Create redirect response to URL

respond_file(self, file_name, headers=None)

• Respond with file content, streaming asynchronously to minimize memory usage
• ⚠️ Security: this method does not restrict file access to any base directory and does not sanitize file_name. If you build file_name from request data (e.g. client.path), an attacker can use .. to escape your web root (path traversal) and read arbitrary files. Always validate the path yourself first: reject any segment equal to .., then join onto a fixed base directory and verify the result stays inside it. Note client.path is already percent-decoded, so %2e%2e and .. look identical to your check.

response_multipart(self, headers=None)

• Create multipart response with headers as dict (for MJPEG streams etc.)

response_multipart_frame(self, data, headers=None, boundary=None)

• Send multipart frame with data and headers

response_multipart_end(self, boundary=None)

• Finish multipart stream

response_stream(self, content_type=None, headers=None, cookies=None)

• Start streaming response without Content-Length
• Default content type: text/event-stream, sets Cache-Control: no-cache
• Returns True on success, False if socket is closed

send_chunk(self, data)

• Send raw data chunk (str or bytes) to stream
• Use for SSE comments (: ping\n\n), custom formats, or non-SSE streaming
• Returns True on success, False if socket is closed

send_event(self, data=None, event=None, event_id=None, retry=None)

• Send formatted SSE event
• data - str or dict/list/tuple/int/float (auto JSON serialized). Multi-line strings split into multiple data: lines
• event - Event type name (client listens via addEventListener())
• event_id - Event ID for client reconnection (Last-Event-ID header)
• retry - Reconnection time in milliseconds
• Returns True on success, False if socket is closed

response_ndjson(self, headers=None, cookies=None)

• Start NDJSON streaming response (application/x-ndjson)
• Thin wrapper over response_stream() with NDJSON content type
• Returns True on success, False if socket is closed

send_ndjson(self, obj)

• Serialize one JSON-serializable value as a single NDJSON line (json.dumps(obj) + '\n')
• obj - any JSON-serializable value (dict/list/str/int/float/bool/None)
• Returns True on success, False if socket is closed

response_stream_end(self)

• End stream and close connection (used for both SSE and NDJSON)

accept_body(self, streaming=False, to_file=None) (event mode only)

• Accept body after EVENT_HEADERS. Call this to start receiving body data.
• streaming=False (default) - Buffer all data, receive only EVENT_COMPLETE
• streaming=True - Receive EVENT_DATA for each chunk, read with read_buffer()
• to_file="/path" - Save body directly to file
• Returns: Number of bytes already waiting in buffer

read_buffer(self) (event mode only)

• Read available data from buffer
• Returns: bytes or None if no data available

accept_websocket(self)

• Accept WebSocket upgrade. Event mode: switches to WS mode. Non-event mode: returns WebSocket object.

ws_send(self, data) (event mode, WebSocket)

• Send WebSocket message. str → text frame, bytes → binary frame.

ws_ping(self, data=b'') (event mode, WebSocket)

• Send WebSocket ping frame

ws_close(self, code=1000, reason='') (event mode, WebSocket)

• Send WebSocket close frame and close connection

Server-Sent Events (SSE)

SSE enables one-way server-to-client streaming over HTTP. The browser's EventSource API handles automatic reconnection.

Basic SSE Stream

import time
import uhttp.server

server = uhttp.server.HttpServer(port=8080)
sse_clients = []

while True:
    client = server.wait(timeout=0.1)

    if client:
        if client.path == '/events':
            if client.response_stream():
                sse_clients.append(client)
        else:
            client.respond("hello")

    # Send events to connected clients
    for sc in list(sse_clients):
        if not sc.send_event({'temp': 23.5}, event='sensor', event_id=1):
            sse_clients.remove(sc)  # Client disconnected

SSE with Reconnection Support

When a client reconnects, the browser sends Last-Event-ID header automatically:

if client.path == '/events':
    last_id = client.header('last-event-id')
    client.response_stream()
    if last_id:
        # Resend missed events from history
        for event in history.since(int(last_id)):
            client.send_event(event['data'], event_id=event['id'])
    sse_clients.append(client)

Keep-alive

Send comments periodically to prevent proxy/load balancer timeouts:

if time.time() - last_ping > 15:
    for sc in list(sse_clients):
        if not sc.send_chunk(': ping\n\n'):
            sse_clients.remove(sc)

SSE Protocol Format

id: 1
event: sensor
retry: 5000
data: {"temp": 23.5}

• Each field on its own line, event terminated by empty line
• data: — event payload (required for client to receive data)
• event: — event type, client listens via addEventListener()
• id: — event ID, sent back as Last-Event-ID on reconnect
• retry: — reconnection delay in milliseconds
• : comment — ignored by client, used for keep-alive pings

NDJSON Streaming

NDJSON (Newline-Delimited JSON, application/x-ndjson) is a minimal one-way server-to-client streaming format: one JSON value per line, separated by \n. Compared to SSE it has no field structure (event:/id:/retry:) and no browser-side reconnect API — just raw JSON records. Useful for bulk data exports, log tailing, internal APIs, and any non-browser HTTP client that consumes a stream of records.

Basic NDJSON Stream

import uhttp.server

server = uhttp.server.HttpServer(port=8080)

while True:
    client = server.wait(timeout=0.1)
    if not client:
        continue

    if client.path == '/export':
        if client.response_ndjson():
            for row in db.iter_rows():
                if not client.send_ndjson(row):
                    break  # client disconnected
            client.response_stream_end()
    else:
        client.respond("hello")

Stream Termination

NDJSON has no in-band end-of-stream marker — the client detects end of stream by TCP connection close (EOF on recv()). Therefore response_stream_end() always closes the connection (no keep-alive).

If you need to signal why the stream ended (e.g. completion vs. server-initiated abort), send a sentinel record on the application level before closing:

client.send_ndjson({'_end': True, 'reason': 'done'})
client.response_stream_end()

Wire Format

{"id":1,"temp":23.5}
{"id":2,"temp":23.7}
{"id":3,"temp":23.6}

• One JSON value per line, terminated by \n
• No leading/trailing wrapping; concatenation of records is the body
• Each send_ndjson() call emits exactly one line
• json.dumps() escapes embedded newlines, so records cannot break the framing

Client Example

import requests, json
with requests.get('http://localhost:8080/export', stream=True) as r:
    for line in r.iter_lines():
        if line:
            record = json.loads(line)
            print(record)

NDJSON vs SSE

Aspect	NDJSON	SSE
Content type	application/x-ndjson	text/event-stream
Per-record metadata	none (just JSON)	event:, id:, retry:
Browser API	none (manual fetch stream)	EventSource w/ auto-reconnect
Multi-line payload	not allowed (one line = one record)	data: repeated per line
End of stream	TCP close	TCP close (or app-level event)
Typical use	bulk export, logs, APIs	live UI updates in browser

WebSocket Support

uHTTP supports WebSocket connections (RFC 6455) in both event mode and non-event mode.

Shared-selector loops: both modes work. In event mode the connection stays the owner; in non-event mode pass the selector to the upgrade — ws = client.accept_websocket(selector=sel) — and the returned WebSocket registers itself as key.data, dispatching through handle_event(fileobj, mask) and next() like every other owner. Without a selector it owns one and you use its blocking wait(timeout).

Security: The server does not validate the Origin header during WebSocket upgrade. To prevent cross-site WebSocket hijacking, validate the Origin header in your application before calling accept_websocket().

Event Mode (non-blocking, multiple connections)

from uhttp.server import (
    HttpServer, EVENT_REQUEST, EVENT_WS_REQUEST,
    EVENT_WS_MESSAGE, EVENT_WS_CHUNK_FIRST,
    EVENT_WS_CHUNK_NEXT, EVENT_WS_CHUNK_LAST,
    EVENT_WS_PING, EVENT_WS_CLOSE
)

server = HttpServer(port=8080, event_mode=True)

while True:
    client = server.wait()
    if not client:
        continue

    if client.event == EVENT_WS_REQUEST:
        client.accept_websocket()

    elif client.event == EVENT_REQUEST:
        client.respond({'status': 'ok'})

    elif client.event == EVENT_WS_MESSAGE:
        data = client.read_buffer()  # bytes
        if client.ws_is_text:
            client.ws_send(data.decode('utf-8'))  # echo as text
        else:
            client.ws_send(data)  # echo as binary

    elif client.event == EVENT_WS_CHUNK_FIRST:
        client.context = {'chunks': [client.read_buffer()]}

    elif client.event == EVENT_WS_CHUNK_NEXT:
        client.context['chunks'].append(client.read_buffer())

    elif client.event == EVENT_WS_CHUNK_LAST:
        client.context['chunks'].append(client.read_buffer())
        # process all chunks...
        client.ws_send('received')

    elif client.event == EVENT_WS_PING:
        pass  # pong sent automatically

    elif client.event == EVENT_WS_CLOSE:
        print("WebSocket closed")

Non-Event Mode (blocking, simple)

from uhttp.server import HttpServer

server = HttpServer(port=8080)

while True:
    client = server.wait()
    if client and client.is_websocket_request:
        ws = client.accept_websocket()  # returns WebSocket object
        while not ws.is_closed:
            msg = ws.recv(timeout=5)
            if msg is not None:
                ws.send(msg)  # echo
    elif client:
        client.respond("hello")

WebSocket API

HttpConnection properties:

• is_websocket_request - True if request is a WebSocket upgrade
• is_websocket - True if connection is in WebSocket mode
• ws_is_text - True if current message is text frame (vs binary)
• ws_message - Ping/close payload (only for EVENT_WS_PING and EVENT_WS_CLOSE)

HttpConnection methods (event mode):

• accept_websocket() - Accept upgrade, switch to WebSocket mode
• read_buffer() - Read message data (returns bytes, use ws_is_text to check type)
• ws_send(data) - Send message (str → text frame, bytes → binary frame)
• ws_ping(data=b'') - Send ping frame
• ws_close(code=1000, reason='') - Close WebSocket connection

WebSocket object (non-event mode):

• recv(timeout=None) - Receive message (blocking). Returns str/bytes/None
• send(data) - Send message (str → text frame, bytes → binary frame)
• ping(data=b'') - Send ping frame
• close(code=1000, reason='') - Close connection
• is_closed - True if connection is closed

Large Message Chunking

Messages larger than MAX_WS_MESSAGE_LENGTH (default 64KB, configurable via max_ws_message_length kwarg) are delivered in chunks via EVENT_WS_CHUNK_FIRST, EVENT_WS_CHUNK_NEXT, EVENT_WS_CHUNK_LAST events. Read chunk data with read_buffer(), check frame type with ws_is_text.

In non-event mode, messages exceeding the limit close the connection with status 1009.

Event Mode

Event mode enables streaming large uploads without buffering entire body in memory.

from uhttp.server import (
    HttpServer, EVENT_REQUEST, EVENT_HEADERS,
    EVENT_DATA, EVENT_COMPLETE, EVENT_ERROR
)

server = HttpServer(port=8080, event_mode=True)

while True:
    client = server.wait()
    if not client:
        continue

    if client.event == EVENT_REQUEST:
        # Small request or GET - handle normally
        client.respond({'status': 'ok'})

    elif client.event == EVENT_HEADERS:
        # Large upload starting - decide how to handle
        client.context = {'total': 0}
        client.accept_body()
        # Read any data that arrived with headers
        data = client.read_buffer()
        if data:
            client.context['total'] += len(data)

    elif client.event == EVENT_DATA:
        # More data arrived
        data = client.read_buffer()
        if data:
            client.context['total'] += len(data)

    elif client.event == EVENT_COMPLETE:
        # Upload finished
        data = client.read_buffer()
        if data:
            client.context['total'] += len(data)
        client.respond({'received': client.context['total']})

    elif client.event == EVENT_ERROR:
        print(f"Error: {client.error}")

For file uploads, use to_file parameter:

elif client.event == EVENT_HEADERS:
    client.accept_body(to_file=f"/uploads/{uuid}.bin")

elif client.event == EVENT_COMPLETE:
    client.respond({'status': 'uploaded'})

Note: Small POST requests where headers and body arrive in the same TCP packet will receive EVENT_REQUEST (not EVENT_HEADERS), since the complete request is already available.

IPv6 Support

Server supports both IPv4 and IPv6:

import uhttp.server

# IPv4 only (default)
server = uhttp.server.HttpServer(address='0.0.0.0', port=80)

# Dual-stack (IPv4 + IPv6)
server = uhttp.server.HttpServer(address='::', port=80)

# IPv6 only
server = uhttp.server.HttpServer(address='::1', port=80)

Development

Running tests

../.venv/bin/pip install -e .
../.venv/bin/python -m unittest discover -v tests/

MicroPython integration tests

Tests run HTTP server on real ESP32 hardware, with test client on PC. Requires mpytool and uhttp-client.

Configuration:

1. WiFi credentials in ~/.config/uhttp/wifi.json:

   {"ssid": "MyWiFi", "password": "secret"}
   

2. Serial port via environment variable or mpytool config:

   # Environment variable
   export MPY_TEST_PORT=/dev/ttyUSB0
   
   # Or mpytool config
   echo "/dev/ttyUSB0" > ~/.config/mpytool/ESP32
   

Run tests:

# Install dependencies
../.venv/bin/pip install uhttp-client mpytool

uhttp-client - for other integration tests mpytool - for mpy_integration tests

# Run tests
MPY_TEST_PORT=/dev/ttyUSB0 ../.venv/bin/python -m unittest tests.test_mpy_integration -v

The tests upload server.py to ESP32, start HTTP server, and send requests from PC.

CI

Tests run automatically via GitHub Actions:

• Unit tests (push + PR): Ubuntu + Windows in parallel, Python 3.10 + 3.14
• MicroPython tests (push only): Self-hosted runner with ESP32, runs after unit tests pass

Expect: 100-continue

Server supports Expect: 100-continue header for large uploads:

Non-event mode: Server automatically sends 100 Continue response when client sends Expect: 100-continue header, then waits for body.

Event mode: Server sends 100 Continue only when application calls accept_body(). This allows rejecting uploads early (e.g., respond with 413 or 401) without accepting body data.

from uhttp.server import HttpServer, EVENT_HEADERS

server = HttpServer(port=8080, event_mode=True)

while True:
    client = server.wait()
    if client and client.event == EVENT_HEADERS:
        if client.content_length > MAX_ALLOWED:
            # No 100 Continue sent - client won't upload
            client.respond({'error': 'too large'}, status=413)
        else:
            client.accept_body()  # Sends 100 Continue, starts receiving