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: すぐに使える高レベルなクライアント/サーバー実装
• 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 を採用
• WebTransport over HTTP/3
  • Sans I/O API と asyncio API の両方を提供
• WebTransport over HTTP/2
  • Sans I/O API と asyncio API の両方を提供
• Python Free-Threading 対応
  • PEP 703 – Making the Global Interpreter Lock Optional in CPython | peps.python.org
  • Python Free-Threading Guide
• クロスプラットフォーム対応
  • 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.