pywebtransport 0.10.0

An async-native WebTransport stack for Python.

PyWebTransport

An 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 fleet management, auto-reconnection, and browser-like navigation.
• Structured Messaging: Send typed Python objects over streams and datagrams using pluggable serializers (JSON, MsgPack, Protobuf).
• Zero-Copy Architecture: End-to-end support for memoryview and buffer protocols to minimize data copying.
• Complete Protocol Implementation: Full support for bidirectional and unidirectional streams, as well as unreliable datagrams.
• Lifecycle and Resource Management: Robust, async context-managed components for handling connections, sessions, streams, and resource cleanup.
• Event-Driven Architecture: A powerful EventEmitter system for decoupled, asynchronous communication between components.
• Type-Safe and Tested: A fully type-annotated API with extensive test coverage (unit, integration, E2E, benchmark) to ensure reliability and maintainability.

Installation

pip install pywebtransport

Quick Start

Server

import asyncio

from pywebtransport import Event, ServerApp, ServerConfig, WebTransportSession, WebTransportStream
from pywebtransport.types import EventType
from pywebtransport.utils import generate_self_signed_cert

generate_self_signed_cert(hostname="localhost")

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


@app.route(path="/")
async def echo_handler(session: WebTransportSession) -> None:
    async def on_datagram(event: Event) -> None:
        if isinstance(event.data, dict):
            payload = event.data.get("data")
            if payload:
                await session.send_datagram(data=b"ECHO: " + payload)

    async def on_stream(event: Event) -> None:
        if isinstance(event.data, dict):
            stream = event.data.get("stream")
            if isinstance(stream, WebTransportStream):
                asyncio.create_task(handle_stream(stream))

    session.events.on(event_type=EventType.DATAGRAM_RECEIVED, handler=on_datagram)
    session.events.on(event_type=EventType.STREAM_OPENED, handler=on_stream)

    try:
        await session.events.wait_for(event_type=EventType.SESSION_CLOSED)
    finally:
        session.events.off(event_type=EventType.DATAGRAM_RECEIVED, handler=on_datagram)
        session.events.off(event_type=EventType.STREAM_OPENED, handler=on_stream)


async def handle_stream(stream: WebTransportStream) -> None:
    try:
        data = await stream.read_all()
        await stream.write_all(data=b"ECHO: " + data)
    except Exception:
        pass


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

Client

import asyncio
import ssl

from pywebtransport import ClientConfig, WebTransportClient
from pywebtransport.types import EventType


async def main() -> None:
    config = ClientConfig(
        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/")

        await session.send_datagram(data=b"Hello, Datagram!")

        event = await session.events.wait_for(event_type=EventType.DATAGRAM_RECEIVED, timeout=5.0)
        if isinstance(event.data, dict):
            print(f"Datagram: {event.data.get('data')!r}")

        stream = await session.create_bidirectional_stream()
        await stream.write_all(data=b"Hello, Stream!")

        response = await stream.read_all()
        print(f"Stream: {response!r}")

        await session.close()


if __name__ == "__main__":
    try:
        asyncio.run(main())
    except KeyboardInterrupt:
        pass

Documentation

• API Reference - Explore the complete API documentation.

Requirements

• Python 3.12+
• TLS 1.3

Contributing

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

Sponsors

Acknowledgments

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

Support

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

License

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