Getting Started with HTTP/3 in Python

Does Python support HTTP/3?

Yes, but not yet natively in the standard library. Python’s http.client and common WSGI frameworks (Flask, Django) are built around HTTP/1.1 or HTTP/2 (sometimes via reverse proxies).

HTTP/3 requires QUIC (QUIC is not TCP, so it’s a big jump). Support in Python is mainly experimental and comes from third-party libraries.

 

Libraries to Work with HTTP/3 in Python

• aioquic → A QUIC and HTTP/3 implementation in Python. You can build both clients and servers with it.
• hypercorn → ASGI server supporting HTTP/1.1, HTTP/2, and experimental HTTP/3 via aioquic.
• httpx → HTTP client for Python. Officially supports HTTP/1.1 and HTTP/2, but HTTP/3 is experimental with aioquic.

 
Perfect 🚀 Let’s sketch out a Python HTTP/3 tutorial outline that mirrors what we did with Go.
For the Golang tutorial 👉 Getting Started with HTTP/3 in Golang: A Practical Guide

 

‘aioquic’ library Explanation:

**aioquic** is a pure-Python QUIC and HTTP/3 library, based on the “bring your own I/O” (sans-I/O) design. It supports QUIC (RFC 9000 & 9369), HTTP/3 (RFC 9114), TLS 1.3, server push, WebSocket, and WebTransport.

The library ships with an example HTTP/3 server (and client) under examples/http3_server.py — They run an ASGI app, serve static files, WebSocket, echo endpoints, and handle HTTP/3 features like server push.

You can refer to the HTTP/3 API docs to understand the low-level classes: H3Connection, send_headers, send_data, etc.

Read the documentation for more information on this library.

 

Server-side Implementation

What you’ll build: a minimalist HTTP/3 server that replies “hello” to GET / and echoes JSON for POST /echo. We’ll cover why each class/argument exists, and how to grow this into something bigger.

Step 0: Prerequisites

• Python: aioquic currently targets modern Python (3.8+ in recent releases)
• TLS: HTTP/3 runs over QUIC + TLS 1.3, so you need a certificate & key (self-signed is fine for local dev). aioquic’s configuration handles loading them.
• Install:

pip install aioquic

 
Step 1: Mental model: where the pieces fit

• QUIC transport: managed by QuicConnection and the asyncio helper serve() for UDP I/O. We don’t touch raw sockets—serve() does that.
• HTTP/3 layer: H3Connection sits on top of QUIC. You feed it QUIC events, and it gives you HTTP/3 events (headers, data, etc.). You respond with send_headers() / send_data().
• Protocol glue: subclass QuicConnectionProtocol and override quic_event_received() to route events into H3Connection .

 
Step 2: Create a self-signed cert (dev only)

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

In production, use a real certificate and proper hostname. QUIC requires TLS 1.3; _aioquic_ will present your certificate via its _QuicConfiguration_.

 
Step 3: Pick the right ALPN

HTTP/3 is negotiated via ALPN. aioquic exposes H3_ALPN (the right tokens for HTTP/3). So you don’t have to guess the strings yourself. We pass that into QuicConfiguration(alpn_protocols=H3_ALPN).

 
Step 4: The minimal but “real” server

Create server.py:

import argparse
import asyncio
import json
import logging
from typing import Dict, Optional
from aioquic.asyncio import QuicConnectionProtocol, serve  # asyncio helpers
from aioquic.h3.connection import H3_ALPN, H3Connection     # HTTP/3 core
from aioquic.h3.events import DataReceived, HeadersReceived, H3Event
from aioquic.quic.configuration import QuicConfiguration
LOG = logging.getLogger("h3server")
class Http3ServerProtocol(QuicConnectionProtocol):
    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self._http: Optional[H3Connection] = None
        # Streams we expect a body for (POST/PUT/PATCH). Value = body buffer.
        self._bodies: Dict[int, bytearray] = {}
        # Remember method/path per stream so DataReceived can route correctly.
        self._req_meta: Dict[int, tuple[str, str]] = {}
    def handle_h3_event(self, event: H3Event) -> None:
        if isinstance(event, HeadersReceived):
            headers = {k.lower(): v for k, v in event.headers}
            method = headers.get(b":method", b"").decode()
            path = headers.get(b":path", b"/").decode()
            sid = event.stream_id
            LOG.info("Headers on stream %d: %s %s", sid, method, path)
            self._req_meta[sid] = (method, path)
            if method in ("POST", "PUT", "PATCH"):
                # We expect a request body.
                self._bodies[sid] = bytearray()
                if event.stream_ended:
                    # Rare case: headers said there's a body, but FIN already set.
                    payload = bytes(self._bodies.pop(sid))
                    self._respond(sid, method, path, payload)
                    self._req_meta.pop(sid, None)
            else:
                # No body expected → respond immediately.
                self._respond(sid, method, path, body=b"")
                # Leave _bodies empty for this stream so DataReceived (empty FIN)
                # gets ignored later.
                self._req_meta.pop(sid, None)
        elif isinstance(event, DataReceived):
            sid = event.stream_id
            # If we never created a body buffer for this stream, we are NOT expecting
            # a body (e.g., GET). Ignore any data/FIN that arrives.
            if sid not in self._bodies:
                return
            # Accumulate body for methods that expect it.
            self._bodies[sid].extend(event.data)
            if event.stream_ended:
                method, path = self._req_meta.get(sid, ("", "/"))
                payload = bytes(self._bodies.pop(sid, bytearray()))
                self._respond(sid, method, path, payload)
                self._req_meta.pop(sid, None)
    # Build and send an HTTP/3 response on a stream
    def _respond(self, stream_id: int, method: Optional[str], path: Optional[str], body: bytes) -> None:
        status = b"200"
        content_type = b"text/plain"
        if method == "GET" and path == "/":
            payload = b"hello from aioquic h3 \n"
        elif method == "POST" and path == "/echo":
            try:
                obj = json.loads(body.decode() or "{}")
                payload = (json.dumps(obj, indent=2) + "\n").encode()
                content_type = b"application/json"
            except Exception:
                payload = body or b"(empty)\n"
                content_type = b"application/octet-stream"
        else:
            payload = b"ok\n"
        headers = [            (b":status", status),
            (b"server", b"aioquic-tutorial"),
            (b"content-type", content_type),
            (b"alt-svc", b'h3=":4433"; ma=3600'),
        ]
        self._http.send_headers(stream_id, headers)
        self._http.send_data(stream_id, payload, end_stream=True)
        LOG.info("Responded on stream %d (%s bytes)", stream_id, len(payload))
async def main():
    parser = argparse.ArgumentParser()
    parser.add_argument("--host", default="::1")
    parser.add_argument("--port", type=int, default=4433)
    parser.add_argument("--certificate", default="cert.pem")
    parser.add_argument("--private-key", dest="private_key", default="key.pem")
    args = parser.parse_args()
    logging.basicConfig(level=logging.INFO)
    # QUIC/TLS configuration
    config = QuicConfiguration(
        is_client=False,
        alpn_protocols=H3_ALPN,
    )
    config.load_cert_chain(args.certificate, args.private_key)
    # 🚀 Start server
    server = await serve(
        host=args.host,
        port=args.port,
        configuration=config,
        create_protocol=Http3ServerProtocol,
    )
    logging.info("HTTP/3 server listening on %s:%d", args.host, args.port)
    # Keep running until cancelled
    try:
        await asyncio.Future()  # run forever
    except KeyboardInterrupt:
        pass
    finally:
        server.close()
if __name__ == "__main__":
    try:
        asyncio.run(main())
    except KeyboardInterrupt:
        pass

 
Why this structure?

• QuicConnectionProtocol gives you the hook quic_event_received()—the right place to turn QUIC events into HTTP/3 events.
• H3Connection is the low-level HTTP/3 engine: you call handle_event() with QUIC events, and it emits HeadersReceived, DataReceived, etc. You respond via send_headers() and send_data().
• QuicConfiguration bundles transport + TLS knobs (ALPN, TLS certs, flow-control, qlog, etc.). We pass H3_ALPN, so the TLS handshake negotiates HTTP/3.
• serve() is the asyncio helper that binds the UDP socket and spins protocol instances per connection.

 
Step 5: Run & test

Start the server:

python server.py --host ::1 --port 4433 --certificate cert.pem --private-key key.pem

After running the above command, you’ll see something like this in the console:

INFO:root:HTTP/3 server listening on ::1:4433

 
Hit it with curl (local, skip cert verification):

curl -v --http3 -k https://localhost:4433/
curl -v --http3 -k https://localhost:4433/echo -d '{"hello": "world"}' \
     -H 'content-type: application/json'

Note: -k ignores TLS validation since we’re using self-signed certs.

 
After executing the first command at the console, you’ll see in the output on the server side:

INFO:root:HTTP/3 server listening on ::1:4433
INFO:quic:[609d9451c3c3ccab] Negotiated protocol version 0x00000001 (VERSION_1)
INFO:quic:[609d9451c3c3ccab] ALPN negotiated protocol h3
INFO:h3server:Handshake complete. ALPN=h3
INFO:h3server:Headers on stream 0: GET /
INFO:h3server:Responded on stream 0 (23 bytes)
INFO:quic:[609d9451c3c3ccab] Connection close received (code 0x0, reason )

And this is on the client side when you call ‘curl’:

hello from aioquic h3

 

Tip: you can add _--qlog qlog.json --secrets secrets.txt_ and then open the QLOG in a viewer, or decrypt PCAPs in Wireshark with TLS secrets. _QuicLogger.to_dict()_ lets you dump traces to JSON.

 
Step 6: Understanding the event flow (so you can extend it)

1. Handshake → you get HandshakeCompleted(alpn_protocol=...). If ALPN is one of H3_ALPN, create H3Connection(self._quic).
2. Requests → HeadersReceived (method/path in pseudo-headers), then optional DataReceived chunks until stream_ended=True.
3. Responses → call send_headers() then send_data(..., end_stream=True).
4. Flush → call self.transmit() to actually push datagrams out (the asyncio helper handles the details).

 
Step 7: Common gotchas (and why)

• Headers must be bytes (names and values). That’s by design in H3Connection.send_headers(...).
• Respond per stream: HTTP/3 is multiplexed; always keep stream_id with your request/response state.
• End the stream: set end_stream=True when you’re done sending data; otherwise, clients wait forever.
• ALPN matters: If your ALPN list doesn’t include HTTP/3 (“h3”), clients won’t negotiate it. Use H3_ALPN.

 
Tipe: How to grow this server

a) Routing & middleware

• Build a tiny router (dict of path → handler) and parse the :method/:path from HeadersReceived.
• Add logging, request IDs, and exception handling around your handler dispatch.

b) Static files & content types

• Look at :path, map to file system, set content-type accordingly, stream file chunks with multiple send_data() calls before end_stream=True.

c) Observability & debugging

• QLOG traces: set config.quic_logger = QuicLogger(); dump to_dict() to JSON to inspect handshakes, streams, congestion, etc.
• TLS secrets: set config.secrets_log_file to decrypt captures in Wireshark.

d) Performance/transport tuning

• Flow control: max_data, max_stream_data in QuicConfiguration.
• DATAGRAM support (for things like WebTransport): set max_datagram_frame_size in QuicConfiguration and handle DatagramReceived on the H3 side.

e) WebTransport (upgrade path)

• Create H3Connection(self._quic, enable_webtransport=True) to receive WebTransport events (unidirectional/bidirectional streams, datagrams). This lets you do low-latency messaging on top of HTTP/3.

f) Production hardening

• Real certificates (ACME/Let’s Encrypt), correct SNI/hostname.
• Run behind a UDP-friendly firewall/load balancer.
• Use a separate H1/H2 server to advertise HTTP/3 via Alt-Svc so that browsers discover your H3 endpoint.
• Consider graceful shutdown: track open streams and close with application errors where appropriate.

 

Client side Implementation

Like the server, the client works in layers:

• QUIC transport: handled by QuicConnection inside QuicConnectionProtocol.
• HTTP/3: H3Connection encodes your requests and parses responses.
• Asyncio helper: instead of serve(), we use connect() to initiate a QUIC connection to a server.

 
Step 1: Mental Model

1. Create QUIC config: QuicConfiguration(is_client=True, alpn_protocols=H3_ALPN).
2. Open QUIC connection: async with connect(...) returns a protocol object.
3. Build request: call h3.send_headers(stream_id, headers) and optionally send_data().
4. Read response events: loop over handle_event(event) until you see the full response.

 
Step 2: Minimal Client Code

Create client.py:

# client_fixed.py
import argparse
import asyncio
import logging
import json
from typing import Optional
from aioquic.asyncio import connect
from aioquic.asyncio.protocol import QuicConnectionProtocol
from aioquic.h3.connection import H3_ALPN, H3Connection
from aioquic.h3.events import HeadersReceived, DataReceived
from aioquic.quic.configuration import QuicConfiguration
from aioquic.quic.events import QuicEvent
LOG = logging.getLogger("h3client")
class H3ClientProtocol(QuicConnectionProtocol):
    """
    QUIC protocol subclass that also manages an H3Connection.
    Important:
      - Accept *args/**kwargs in __init__ because aioquic.connect() will call
        the factory with signature: create_protocol(connection, stream_handler=...)
      - Call super().__init__(*args, **kwargs) so the base class sets up self._quic.
    """
    def __init__(self, *args, **kwargs):
        # accept arbitrary args (connection, stream_handler=...) from connect()
        super().__init__(*args, **kwargs)
        # H3 layer bound to the underlying QUIC connection
        self._http = H3Connection(self._quic)
        # per-request state (this minimal example supports one outstanding request)
        self._stream_id: Optional[int] = None
        self._response_headers = None
        self._response_body = bytearray()
        self._done = asyncio.Event()
    def request(self, method: str, path: str, body: bytes = b""):
        """
        Build and send a simple request on a new bidirectional stream.
        """
        # get a client-initiated bidirectional stream id
        self._stream_id = self._quic.get_next_available_stream_id()
        headers = [            (b":method", method.encode()),
            (b":scheme", b"https"),
            (b":authority", b"localhost"),
            (b":path", path.encode()),
        ]
        # send headers then body; end_stream=True to finish the request
        self._http.send_headers(self._stream_id, headers)
        self._http.send_data(self._stream_id, body or b"", end_stream=True)
        # make sure bytes get pushed out
        self.transmit()
    def quic_event_received(self, event: QuicEvent):
        """
        Called by aioquic when QUIC-level events arrive.
        Hand them to H3Connection, and handle resulting H3 events.
        """
        for h3_event in self._http.handle_event(event):
            if isinstance(h3_event, HeadersReceived):
                self._response_headers = h3_event.headers
            elif isinstance(h3_event, DataReceived):
                self._response_body.extend(h3_event.data)
                if h3_event.stream_ended:
                    self._done.set()
    async def wait_response(self):
        """Wait for response completion (stream ended) and return headers+body."""
        await self._done.wait()
        return self._response_headers, bytes(self._response_body)
async def main():
    parser = argparse.ArgumentParser()
    parser.add_argument("--host", default="localhost")
    parser.add_argument("--port", type=int, default=4433)
    parser.add_argument("--path", default="/")
    parser.add_argument("--insecure", action="store_true", default=True)
    args = parser.parse_args()
    logging.basicConfig(level=logging.INFO)
    config = QuicConfiguration(is_client=True, alpn_protocols=H3_ALPN)
    if args.insecure:
        # Only for dev/testing
        config.verify_mode = False
    # Use create_protocol=H3ClientProtocol (aioquic will call it as factory)
    async with connect(
        args.host,
        args.port,
        configuration=config,
        create_protocol=H3ClientProtocol,
    ) as protocol:
        # 'protocol' is an instance of H3ClientProtocol (the factory's return)
        protocol.request("GET", args.path)
        headers, body = await protocol.wait_response()
        LOG.info("Response headers: %s", headers)
        LOG.info("Response body:\n%s", body.decode(errors="ignore"))
if __name__ == "__main__":
    asyncio.run(main())

🔍 Breaking down **H3ClientProtocol**

class H3ClientProtocol(QuicConnectionProtocol):
    """
    QUIC protocol subclass that also manages an H3Connection.
    Important:
      - Accept *args/**kwargs in __init__ because aioquic.connect() will call
        the factory with signature: create_protocol(connection, stream_handler=...)
      - Call super().__init__(*args, **kwargs) so the base class sets up self._quic.
    """

 
This is a thin wrapper around two moving parts:

1. QUIC connection (self._quic) → the transport.
2. H3Connection (self._http) → the HTTP/3 layer that speaks in requests/responses.

The job of this class is:

• Encode an HTTP/3 request (method, path, headers, body).
• Feed QUIC events into H3Connection.
• Collect HTTP/3 response headers and body.
• Expose a nice wait_response() method to the caller.

**__init__**

def __init__(self, *args, **kwargs):
        # accept arbitrary args (connection, stream_handler=...) from connect()
        super().__init__(*args, **kwargs)
        # H3 layer bound to the underlying QUIC connection
        self._http = H3Connection(self._quic)
        # per-request state (this minimal example supports one outstanding request)
        self._stream_id: Optional[int] = None
        self._response_headers = None
        self._response_body = bytearray()
        self._done = asyncio.Event()

• self._quic → reference to the live QUIC connection from aioquic.connect().
• self._http = H3Connection(self._quic) → attaches an HTTP/3 session on top of QUIC.
• self._stream_id → stores the ID of the request stream (each request lives on one bidirectional stream).
• self._response_headers → will hold the [(b':status', b'200'), ...] list when the server replies.
• self._response_body → a bytearray buffer where we accumulate DataReceived chunks.
• self._done → an asyncio.Event we use to signal “the full response is here.”

👉 Why? HTTP/3 is multiplexed: you may have multiple streams in flight, so you need per-stream tracking. This class only handles one stream at a time (a minimal example).

 
**request()**

    def request(self, method: str, path: str, body: bytes = b""):
        """
        Build and send a simple request on a new bidirectional stream.
        """
        # get a client-initiated bidirectional stream id
        self._stream_id = self._quic.get_next_available_stream_id()
        headers = [            (b":method", method.encode()),
            (b":scheme", b"https"),
            (b":authority", b"localhost"),
            (b":path", path.encode()),
        ]
        # send headers then body; end_stream=True to finish the request
        self._http.send_headers(self._stream_id, headers)
        self._http.send_data(self._stream_id, body or b"", end_stream=True)
        # make sure bytes get pushed out
        self.transmit()

• First, we open a new stream: get_next_available_stream_id() asks QUIC for the next free stream number.
• Then, we build the mandatory HTTP/3 pseudo-headers:
  • :method = GET/POST/etc.
  • :scheme = https (QUIC always uses TLS).
  • :authority = host/port (like Host: in HTTP/1.1).
  • :path = request path (e.g. /echo).
• Send the headers into the HTTP/3 engine with send_headers().
• Send a body (if any) with send_data(..., end_stream=True) to mark the request finished.

👉 Why? HTTP/3 requires the pseudo-headers exactly like HTTP/2, otherwise the request is invalid. Ending the stream (end_stream=True) is important; otherwise, the server keeps waiting for more data.

 
**quic_event_received()**

    def quic_event_received(self, event: QuicEvent):
        """
        Called by aioquic when QUIC-level events arrive.
        Hand them to H3Connection, and handle the resulting H3 events.
        """
        for h3_event in self._http.handle_event(event):
            if isinstance(h3_event, HeadersReceived):
                self._response_headers = h3_event.headers
            elif isinstance(h3_event, DataReceived):
                self._response_body.extend(h3_event.data)
                if h3_event.stream_ended:
                    self._done.set()

This method connects the layers:

• QUIC emits events (like “data came in on stream X”).
• We feed them into self._http.handle_event(event).
• That yields HTTP/3 events such as:
  • HeadersReceived: first part of the response → save headers.
  • DataReceived: chunk of the response body → append to buffer.
• If event.stream_ended is true, it means the server finished sending → we set the asyncio.Event so wait_response() unblocks.

👉 Why? HTTP/3 is an event-driven protocol. We must translate raw QUIC events into meaningful HTTP/3 objects and handle them incrementally.

 
**wait_response()**

async def wait_response(self):
    await self._done.wait()
    return self._response_headers, bytes(self._response_body)

• This is a synchronization point for the application.
• The caller awaits here until the _done event is set (when the stream ended).
• Then we return the headers + the full body (as immutable bytes).

👉 Why? In real HTTP/3 clients, you might want streaming (process chunks as they arrive), but here we keep it simple: collect everything, then return.

 

🔑 Why do we need this class at all?

We could interact with H3Connection directly in the main loop, but wrapping it in a class gives us:

• Encapsulation: request(), wait_response() are nicer than juggling stream_ids and events manually.
• State tracking: keeps stream_id, buffers, and headers together.
• Extensibility: later, we can add:
  • support for multiple streams at once,
  • timeouts,
  • custom error handling,
  • QLOG logging,
  • streaming request/response APIs.

 
Step 3: Test it

Run your server:

python h3_server.py --host :: --port 4433 --certificate cert.pem --private-key key.pem

 
Run the client:

python h3_client.py --host localhost --port 4433 --path /

You should see logs like:

INFO:quic:[110acafccf372968] Negotiated protocol version 0x00000001 (VERSION_1)
INFO:quic:[110acafccf372968] ALPN negotiated protocol h3
INFO:h3client:Response headers: [(b':status', b'200'), (b'server', b'aioquic-tutorial'), (b'content-type', b'text/plain'), (b'alt-svc', b'h3=":4433"; ma=3600')]
INFO:h3client:Response body:
hello from aioquic h3 
INFO:quic:[110acafccf372968] Connection close sent (code 0x0, reason )

 
Step 4: Improvements

• Add POST with --data flag to send a body.
• Support multiple requests (reuse connection, open multiple streams).
• Validate certs (skip --insecure in production).
• Collect timing info (request start/end).
• Dump QLOG on the client side for debugging.