pywebtransport 0.6.1

The canonical, async-native WebTransport stack for Python.

PyWebTransport

The canonical, async-native WebTransport stack for Python.

Features

• Full Async Support: Built from the ground up on asyncio for high-performance, non-blocking I/O.
• High-Level Frameworks: Includes a ServerApp with routing and middleware, and a versatile WebTransportClient with helpers for pooling, auto-reconnection, and proxying.
• Complete Protocol Implementation: Full support for bidirectional and unidirectional streams, as well as unreliable datagrams.
• Structured Messaging: Pluggable JSON, MsgPack, and Protobuf serializers for sending and receiving structured data objects over streams and datagrams.
• Lifecycle and Resource Management: Robust, async context-managed components for handling connections, sessions, streams, and monitoring.
• Event-Driven Architecture: A powerful EventEmitter and EventBus system for decoupled, asynchronous communication between components.
• Type-Safe and Tested: A fully type-annotated API with extensive test coverage (unit, integration, E2E) to ensure reliability and maintainability.

Installation

pip install pywebtransport

For more detailed instructions, including virtual environments and platform-specific notes, see the Installation Guide.

Quick Start

Server

# server.py
import asyncio

from pywebtransport import (
    ConnectionError,
    ServerApp,
    ServerConfig,
    SessionError,
    WebTransportSession,
    WebTransportStream,
)
from pywebtransport.utils import generate_self_signed_cert

generate_self_signed_cert(hostname="localhost")

app = ServerApp(
    config=ServerConfig.create(
        certfile="localhost.crt",
        keyfile="localhost.key",
        initial_max_data=1024 * 1024,
        initial_max_streams_bidi=10,
    )
)


async def handle_datagrams(session: WebTransportSession) -> None:
    try:
        datagram_transport = await session.datagrams
        while True:
            data = await datagram_transport.receive()
            await datagram_transport.send(data=b"ECHO: " + data)
    except (ConnectionError, SessionError, asyncio.CancelledError):
        pass


async def handle_streams(session: WebTransportSession) -> None:
    try:
        async for stream in session.incoming_streams():
            if isinstance(stream, WebTransportStream):
                data = await stream.read_all()
                await stream.write_all(data=b"ECHO: " + data)
    except (ConnectionError, SessionError, asyncio.CancelledError):
        pass


@app.route(path="/")
async def echo_handler(session: WebTransportSession) -> None:
    datagram_task = asyncio.create_task(handle_datagrams(session))
    stream_task = asyncio.create_task(handle_streams(session))
    try:
        await session.wait_closed()
    finally:
        datagram_task.cancel()
        stream_task.cancel()


if __name__ == "__main__":
    app.run(host="127.0.0.1", port=4433)

Client

# client.py
import asyncio
import ssl

from pywebtransport import ClientConfig, WebTransportClient


async def main() -> None:
    config = ClientConfig.create(
        verify_mode=ssl.CERT_NONE,
        initial_max_data=1024 * 1024,
        initial_max_streams_bidi=10,
    )

    async with WebTransportClient(config=config) as client:
        session = await client.connect(url="https://127.0.0.1:4433/")

        print("Connection established. Testing datagrams...")
        datagram_transport = await session.datagrams
        await datagram_transport.send(data=b"Hello, Datagram!")
        response = await datagram_transport.receive()
        print(f"Datagram echo: {response!r}\n")

        print("Testing streams...")
        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

Documentation

• Installation Guide - In-depth setup and installation guide.
• Quick Start - A 5-minute tutorial to get started.
• API Reference - Complete API documentation.

Requirements

• Python 3.11+
• asyncio support
• TLS 1.3

Dependencies:

• aioquic >= 1.2.0
• cryptography >= 45.0.4

Contributing

Contributions are welcome! Please read our Contributing Guide for details on the development setup, testing, and pull request process.

Development Setup:

git clone https://github.com/lemonsterfy/pywebtransport.git
cd pywebtransport
pip install -r dev-requirements.txt
pip install -e .
tox

License

This project is licensed under the MIT License - see the LICENSE file for details.

Acknowledgments

• aioquic for the underlying QUIC protocol implementation.
• WebTransport Working Group for standardizing the protocol.

Support

• Issues: GitHub Issues
• Discussions: GitHub Discussions
• Email: [e-mail]