Hello everyone,
I’m excited to announce the release of PyWebTransport, a new, open-source library that provides a production-ready, async-native WebTransport stack for Python.
Motivation
For years, WebSockets have been the standard for real-time, bidirectional communication on the web. However, they are built on TCP, which can suffer from head-of-line blocking, and they lack native support for unreliable datagrams or multiple independent streams over a single connection.
PyWebTransport is designed to solve these problems by leveraging the modern QUIC protocol. It offers a powerful alternative for building next-generation, high-performance network applications in Python, providing multiple streams, unidirectional streams, and unreliable datagrams out of the box.
The goal is to provide the Python ecosystem with a canonical, robust, and easy-to-use implementation of this modern protocol.
Key Features & Architecture
The library is built from the ground up on asyncio and is designed with production use in mind.
-
Async-Native Design: A fully asynchronous API for high-performance, non-blocking I/O.
-
Purpose-Built H3 Engine: While it uses the excellent
aioquiclibrary for the QUIC transport layer, it features a purpose-built HTTP/3 engine that is strictly focused on the needs of WebTransport, resulting in a simpler and more maintainable implementation. -
High-Level Frameworks: Includes a
ServerAppwith path-based routing and middleware, and a versatileWebTransportClientwith helpers for connection pooling and an optional auto-reconnect strategy with exponential backoff. -
Structured Messaging: A pluggable serialization layer with out-of-the-box support for JSON, MsgPack, and Protobuf, allowing you to send and receive structured Python objects directly.
-
Type-Safe and Tested: A fully type-annotated API with extensive test coverage (>95%) to ensure reliability.
A Quick Look at the API
Here is a brief example of a simple echo server and client.
Server
import asyncio
from pywebtransport import ServerApp, ServerConfig, WebTransportSession
from pywebtransport.utils import generate_self_signed_cert
# In a real application, you would use a proper certificate
generate_self_signed_cert(hostname="localhost")
app = ServerApp(
config=ServerConfig.create(
certfile="localhost.crt",
keyfile="localhost.key",
)
)
@app.route("/")
async def echo_handler(session: WebTransportSession) -> None:
try:
# This example handles one datagram and one stream, then closes.
datagrams = await session.datagrams
data = await datagrams.receive()
await datagrams.send(b"ECHO: " + data)
stream = await session.accept_stream()
data = await stream.read_all()
await stream.write_all(b"ECHO: " + data)
await session.wait_closed()
except asyncio.CancelledError:
pass
if __name__ == "__main__":
app.run(host="127.0.0.1", port=4433)
Client
import asyncio
import ssl
from pywebtransport import ClientConfig, WebTransportClient
async def main() -> None:
config = ClientConfig.create(verify_mode=ssl.CERT_NONE)
async with WebTransportClient(config=config) as client:
session = await client.connect(url="https://127.0.0.1:4433/")
datagrams = await session.datagrams
await datagrams.send(data=b"Hello, Datagram!")
response = await datagrams.receive()
print(f"Datagram echo: {response!r}")
stream = await session.create_bidirectional_stream()
await stream.write_all(data=b"Hello, Stream!")
response = await stream.read_all()
print(f"Stream echo: {response!r}")
await session.close()
if __name__ == "__main__":
try:
asyncio.run(main())
except KeyboardInterrupt:
pass
Links
- GitHub (Source Code & Issues): GitHub - lemonsterfy/pywebtransport: An async-native WebTransport stack for Python.
- Documentation (API Reference & Guides): https://pywebtransport.readthedocs.io
- PyPI (Package): Client Challenge
The library is still pre-1.0, but the API is stabilizing. I welcome all feedback, suggestions, and contributions from the community as we work towards a full 1.0 release.