webtransport-py

About Shiguredo's open source software

We will not respond to PRs or issues that have not been discussed on Discord. Also, Discord is only available in Japanese.

Please read https://github.com/shiguredo/oss/blob/master/README.en.md before use.

時雨堂のオープンソースソフトウェアについて

利用前に https://github.com/shiguredo/oss をお読みください。

webtransport-py について

webtransport-py は Sans I/O アーキテクチャを採用した WebTransport の Python ライブラリです。WebTransport over HTTP/3 と WebTransport over HTTP/2 の両方に対応しています。

また、WebTransport だけでなく QUIC、HTTP/3、HTTP/2 を単体のプロトコルとしても利用できます。asyncio、スレッド、独自のイベントループなど、任意の I/O フレームワークと組み合わせて利用できます。おまけとして QMux も Sans I/O API と asyncio API の両方を提供しています。

特徴

• Sans I/O アーキテクチャ
  • I/O 処理をライブラリ外部で制御可能
  • 任意のイベントループやフレームワークと統合可能
  • Sans I/O
• 二層 API 設計
  • Sans I/O API: プロトコル処理のみを提供する低レベル API
  • asyncio API: すぐに使える高レベルなクライアント/サーバー実装
• WebTransport over HTTP/3
  • Sans I/O API と asyncio API の両方を提供
• WebTransport over HTTP/2
  • Sans I/O API と asyncio API の両方を提供
• QUIC
  • Sans I/O API と asyncio API の両方を提供
  • 双方向/単方向ストリーム
  • QUIC DATAGRAM
  • 0-RTT / Session Resumption
  • 証明書のカスタム検証
  • Connection Migration
  • ngtcp2 を採用
• HTTP/3
  • Sans I/O API と asyncio API の両方を提供
  • nghttp3 を採用
• HTTP/2
  • Sans I/O API と asyncio API の両方を提供
  • nghttp2-v2 を採用
• QMux
  • 任意の双方向バイトストリーム上でストリーム多重化を提供
  • Sans I/O API と asyncio API の両方を提供
  • dwnx を採用
  • DATAGRAM は未対応
• 依存ライブラリは deps.json で特定のタグ / コミットに固定する
  • ngtcp2 / nghttp3 / dwnx は上流ブランチの特定コミット (ref)、nghttp2-v2 / AWS-LC はタグ (tag) で固定する
  • 更新時は deps.json を書き換える (ビルドキャッシュのキーも deps.json の内容に連動する)
• Python Free-Threading 対応
  • PEP 703 – Making the Global Interpreter Lock Optional in CPython | peps.python.org
  • Python Free-Threading Guide
  • 同一オブジェクトへの並行アクセスはオブジェクト単位の排他で保護する (接続・セッション系クラスの公開メソッドが対象。生成系 static・設定・イベント等の値オブジェクトは対象外)。Config は単一スレッドで構築し、共有後の並行書き換えは行わないこと。quic.Config の検証コールバック内で同一 quic.Connection のメソッドを呼ばないこと
• クロスプラットフォーム対応
  • Ubuntu x86_64 / arm64
  • macOS arm64

インストール

uv add webtransport-py

使い方

WebTransport

HTTPVersion で HTTP/2 (TCP + TLS) と HTTP/3 (UDP + QUIC) を選ぶ。既定は HTTP/3。

クライアント

import asyncio

from webtransport import Client, HTTPVersion, WebTransportError


async def main() -> None:
    client = Client(
        url="https://localhost:4433/webtransport",
        http_version=HTTPVersion.HTTP3,
        verify_peer=False,
    )

    async def on_datagram(data: bytes) -> None:
        print(f"データグラム受信: {data}")

    client.on_datagram(on_datagram)

    try:
        await client.connect()
    except WebTransportError as exc:
        print(f"接続失敗: {exc}")
        await client.close()
        return

    await client.send_datagram(b"Hello, WebTransport!")

    try:
        await asyncio.wait_for(client.run(), timeout=5.0)
    except TimeoutError:
        pass

    await client.close()


asyncio.run(main())

サーバー

import asyncio

from webtransport import Client, HTTPVersion, Server, Session


async def main() -> None:
    server = Server(
        host="0.0.0.0",
        port=4433,
        http_version=HTTPVersion.HTTP3,
        certfile="cert.pem",
        keyfile="key.pem",
    )

    async def on_datagram(session: Session, data: bytes) -> None:
        # セッションハンドル経由で返す
        await session.send_datagram(data)

    server.on_datagram(on_datagram)

    async with server:
        await server.run()


asyncio.run(main())

http_version に HTTPVersion.HTTP2 を渡すと WebTransport over HTTP/2 になる (certfile / keyfile が必須)。プロトコル固有の API は client.h3 / client.h2 / server.h3 / server.h2 から呼ぶ。

低レベル (Sans-IO) API と全 API のリファレンスは skills/webtransport-py/SKILL.md を参照。

QUIC

クライアント

import asyncio

from webtransport import quic


async def main() -> None:
    client = quic.Client(
        host="localhost",
        port=4433,
        verify_peer=False,
    )

    if not await client.connect():
        print("接続失敗")
        return

    # 双方向ストリームを開いてデータ送信 (FIN でストリームを閉じる)
    stream_id = await client.open_stream(bidirectional=True)
    await client.send_stream_data(stream_id, b"Hello, QUIC!", fin=True)

    # サーバーからのエコーを FIN まで受信する
    try:
        data, _ = await client.recv_stream_data(stream_id, timeout=5.0)
        print(f"データ受信: {data}")
    except TimeoutError:
        print("受信タイムアウト")

    await client.close()


if __name__ == "__main__":
    asyncio.run(main())

サーバー

import asyncio

from webtransport import quic


async def main() -> None:
    server = quic.Server(
        host="0.0.0.0",
        port=4433,
        certfile="cert.pem",
        keyfile="key.pem",
    )

    async def on_handshake_completed(addr: tuple[str, int]) -> None:
        print(f"ハンドシェイク完了: {addr}")

    async def on_stream_data(
        stream_id: int,
        data: bytes,
        fin: bool,
        addr: tuple[str, int],
    ) -> None:
        print(f"データ受信: {data}")
        # エコーバック
        await server.send_stream_data(addr, stream_id, data, fin)

    server.on_handshake_completed(on_handshake_completed)
    server.on_stream_data(on_stream_data)

    async with server:
        print(f"サーバー開始: {server.host}:{server.actual_port}")
        await server.run()


if __name__ == "__main__":
    asyncio.run(main())

HTTP/3

クライアント

import asyncio

from webtransport import http3


async def main() -> None:
    client = http3.Client(
        host="localhost",
        port=4433,
        verify_peer=False,
    )

    async def on_headers(stream_id: int, headers: list[tuple[str, str]]) -> None:
        print(f"ヘッダー受信 (stream_id={stream_id}): {headers}")

    async def on_data(stream_id: int, data: bytes) -> None:
        print(f"データ受信: {data}")

    client.on_headers(on_headers)
    client.on_data(on_data)

    await client.connect()
    await client.request("GET", "/")

    try:
        await asyncio.wait_for(client.run(), timeout=5.0)
    except TimeoutError:
        pass

    await client.close()


if __name__ == "__main__":
    asyncio.run(main())

サーバー

import asyncio

from webtransport import http3


async def main() -> None:
    server = http3.Server(
        host="0.0.0.0",
        port=4433,
        certfile="cert.pem",
        keyfile="key.pem",
    )

    async def on_request(
        stream_id: int,
        headers: list[tuple[str, str]],
        addr: tuple[str, int],
    ) -> None:
        print(f"リクエスト受信 (stream_id={stream_id}): {headers}")

    async def on_stream_end(stream_id: int, addr: tuple[str, int]) -> None:
        response_headers: list[tuple[str, str]] = [
            (":status", "200"),
            ("content-type", "text/plain"),
        ]
        await server.submit_response(addr, stream_id, response_headers)
        await server.send_data(addr, stream_id, b"Hello from HTTP/3 server!", fin=True)

    server.on_request(on_request)
    server.on_stream_end(on_stream_end)

    async with server:
        print("Ctrl+C で終了")
        try:
            await server.run()
        except KeyboardInterrupt:
            pass


if __name__ == "__main__":
    asyncio.run(main())

HTTP/2

クライアント

import asyncio

from webtransport import http2


async def main() -> None:
    client = http2.Client(
        host="localhost",
        port=8443,
        verify_peer=False,
    )

    async def on_headers(stream_id: int, headers: list[tuple[str, str]]) -> None:
        print(f"ヘッダー受信 (stream_id={stream_id}): {headers}")

    async def on_data(stream_id: int, data: bytes) -> None:
        print(f"データ受信: {data}")

    client.on_headers(on_headers)
    client.on_data(on_data)

    await client.connect()
    await client.request("GET", "/")

    try:
        await asyncio.wait_for(client.run(), timeout=5.0)
    except TimeoutError:
        pass

    await client.close()


if __name__ == "__main__":
    asyncio.run(main())

サーバー

import asyncio

from webtransport import http2


async def main() -> None:
    server = http2.Server(
        host="0.0.0.0",
        port=8443,
        certfile="cert.pem",
        keyfile="key.pem",
    )

    async def on_request(
        stream_id: int,
        headers: list[tuple[str, str]],
        response_writer: http2.ResponseWriter,
    ) -> None:
        print(f"リクエスト受信 (stream_id={stream_id}): {headers}")

    async def on_stream_end(
        stream_id: int,
        response_writer: http2.ResponseWriter,
    ) -> None:
        response_headers: list[tuple[str, str]] = [
            (":status", "200"),
            ("content-type", "text/plain"),
        ]
        await response_writer.send_headers(stream_id, response_headers)
        await response_writer.send_data(
            stream_id,
            b"Hello from HTTP/2 server!",
            end_stream=True,
        )

    server.on_request(on_request)
    server.on_stream_end(on_stream_end)

    async with server:
        print("Ctrl+C で終了")
        try:
            await server.run()
        except KeyboardInterrupt:
            pass


if __name__ == "__main__":
    asyncio.run(main())

QMux

QMux (draft-ietf-quic-qmux) は任意の双方向バイトストリーム上でストリーム多重化を提供するプロトコルです。

クライアント

import asyncio

from webtransport import qmux


async def main() -> None:
    client = qmux.Client("127.0.0.1", 4433)

    async def on_stream_data(stream_id: int, data: bytes, fin: bool) -> None:
        print(f"ストリームデータ受信: {data}")

    client.on_stream_data(on_stream_data)

    await client.connect()
    stream_id = await client.open_stream()
    await client.send_stream_data(stream_id, b"Hello, QMux!", fin=True)
    await asyncio.wait_for(client.run(), timeout=5.0)
    await client.close()


asyncio.run(main())

サーバー

import asyncio

from webtransport import qmux


async def main() -> None:
    server = qmux.Server("127.0.0.1", 4433)

    async def on_stream_data(
        session: qmux.Session,
        stream_id: int,
        data: bytes,
        fin: bool,
    ) -> None:
        # 双方向ストリームのみエコーする
        if stream_id % 4 == 0:
            await session.send_stream_data(stream_id, data, fin=fin)

    server.on_stream_data(on_stream_data)

    await server.start()
    await server.run()


asyncio.run(main())

ssl に SSLContext を渡すと TLS 上で動きます (alpn_protocols が必須です)。

from webtransport import qmux

config = qmux.Config()
client = qmux.Connection.create_client(config)
server = qmux.Connection.create_server(config)

# 受け取ったバイト列を渡し、送信すべきレコードを取り出す
client.receive(record_from_peer)
record = client.pending_record

レコード (varint の長さ + QUIC フレーム列) をバイト列として入出力します。トランスポートパラメータは帯域内で交換するため、TLS が無くても 2 つの接続をバイト列で直結すればハンドシェイクできます。

Python

• 3.14
• 3.14t
• 3.15
• 3.15t

プラットフォーム

• Ubuntu 26.04 LTS x86_64
• Ubuntu 26.04 LTS arm64
• Ubuntu 24.04 LTS x86_64
• Ubuntu 24.04 LTS arm64
• macOS 26 arm64

リリースビルド

make wheel

開発ビルド

make develop

テスト

uv sync
make test

サンプル

examples/ ディレクトリにサンプルコードがあります。

第三者ライセンス

本プロジェクトは ngtcp2 / nghttp3 / nghttp2-v2 / dwnx / AWS-LC / nanobind (同梱の tsl::robin_map を含む) を静的リンクしています。各ライセンス全文は THIRD_PARTY_LICENSES.md を参照してください。

ライセンス

Apache License 2.0

Copyright 2026 Shiguredo Inc.

Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at

    http://www.apache.org/licenses/LICENSE-2.0

Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.