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 フレームワークと組み合わせて利用できます。

特徴

• 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 を採用
• 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 over HTTP/3

サーバー

import asyncio

from webtransport import h3


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

    async def on_session_ready(session_id: int, addr: tuple[str, int]) -> None:
        print(f"セッション確立: {session_id} from {addr}")

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

    async def on_datagram(session_id: int, data: bytes, addr: tuple[str, int]) -> None:
        print(f"データグラム受信: {data}")
        # エコーバック
        await server.send_datagram(addr, session_id, data)

    server.on_session_ready(on_session_ready)
    server.on_stream_data(on_stream_data)
    server.on_datagram(on_datagram)

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


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

クライアント

import asyncio

from webtransport import h3


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

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

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

    client.on_stream_data(on_stream_data)
    client.on_datagram(on_datagram)

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

    # ストリームでデータ送信
    stream_id = await client.open_stream()
    await client.send_stream_data(stream_id, b"Hello via stream!")

    # データグラムでデータ送信
    await client.send_datagram(b"Hello via datagram!")

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

    await client.close()


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

WebTransport over HTTP/2

サーバー

import asyncio

from webtransport import h2


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

    async def on_session_ready(session_writer: h2.SessionWriter) -> None:
        print(f"セッション確立: {session_writer.session_id}")

    async def on_stream_data(
        stream_id: int,
        data: bytes,
        session_writer: h2.SessionWriter,
    ) -> None:
        print(f"データ受信: {data}")
        # エコーバック
        await session_writer.send_stream_data(stream_id, data)

    async def on_datagram(data: bytes, session_writer: h2.SessionWriter) -> None:
        print(f"データグラム受信: {data}")
        # エコーバック
        await session_writer.send_datagram(data)

    server.on_session_ready(on_session_ready)
    server.on_stream_data(on_stream_data)
    server.on_datagram(on_datagram)

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


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

クライアント

import asyncio

from webtransport import h2


async def main() -> None:
    client = h2.Client(
        url="https://localhost:8443/webtransport",
        verify_peer=False,
    )

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

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

    client.on_stream_data(on_stream_data)
    client.on_datagram(on_datagram)

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

    # ストリームでデータ送信
    stream_id = await client.open_stream()
    await client.send_stream_data(stream_id, b"Hello via stream!")

    # データグラムでデータ送信
    await client.send_datagram(b"Hello via datagram!")

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

    await client.close()


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

QUIC

サーバー

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())

クライアント

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())

Python

• 3.14
• 3.14t

プラットフォーム

• 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 / AWS-LC を静的リンクしています。各ライセンス全文は 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.