# PyWebTransport
[](https://badge.fury.io/py/pywebtransport)
[](https://pypi.org/project/pywebtransport/)
[](https://opensource.org/licenses/MIT)
[](https://github.com/lemonsterfy/pywebtransport/actions)
[](https://codecov.io/gh/lemonsterfy/pywebtransport)
[](https://docs.pywebtransport.org/en/latest/)
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.
- **Advanced Messaging**: Built-in managers for Pub/Sub and RPC (JSON-RPC 2.0 compliant), plus pluggable serializers (`JSON`, `MsgPack`, `Protobuf`) for structured data.
- **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, monitoring, pooling, and resource management.
- **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) to ensure reliability and maintainability.
## Installation
```bash
pip install pywebtransport
```
For more detailed instructions, including virtual environments and platform-specific notes, see the [Installation Guide](docs/installation.md).
## Quick Start
### Server
```python
# 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(
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.create_datagram_transport()
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
```python
# client.py
import asyncio
import ssl
from pywebtransport import ClientConfig, WebTransportClient
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/")
print("Connection established. Testing datagrams...")
datagram_transport = await session.create_datagram_transport()
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](docs/installation.md)** - Learn how to install the library.
- **[Quick Start](docs/quickstart.md)** - Discover how to get started with basic client and server setup.
- **[API Reference](docs/api-reference/)** - Explore the complete API documentation.
## Requirements
- Python 3.11+
- asyncio support
- TLS 1.3
**Dependencies:**
- aioquic >= 1.3.0
- cryptography >= 45.0.4
## Contributing
Contributions are welcome! Please read our [Contributing Guide](CONTRIBUTING.md) for details on the development setup, testing, and pull request process.
**Development Setup:**
```bash
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](LICENSE) file for details.
## Acknowledgments
- <a href="https://www.fastly.com/" target="_blank" rel="noopener noreferrer">
<img src="https://upload.wikimedia.org/wikipedia/commons/8/8a/Fastly_logo.svg" alt="Fastly" width="80">
</a>
for providing critical infrastructure and services.
- [aioquic](https://github.com/aiortc/aioquic) for the underlying QUIC protocol implementation.
- [WebTransport Working Group](https://datatracker.ietf.org/wg/webtrans/) for defining and standardizing the WebTransport protocol.
## Support
- **Issues**: [GitHub Issues](https://github.com/lemonsterfy/pywebtransport/issues)
- **Discussions**: [GitHub Discussions](https://github.com/lemonsterfy/pywebtransport/discussions)
- **Email**: lemonsterfy@gmail.com
Raw data
{
"_id": null,
"home_page": null,
"name": "pywebtransport",
"maintainer": null,
"docs_url": null,
"requires_python": ">=3.11",
"maintainer_email": null,
"keywords": "async, http3, networking, protocol, quic, webtransport",
"author": null,
"author_email": "lemonsterfy <lemonsterfy@gmail.com>",
"download_url": "https://files.pythonhosted.org/packages/d4/ae/1314a4a409f9b1ffa2963ef285f3507829581014ef9ecd61737bb32aaa99/pywebtransport-0.8.1.tar.gz",
"platform": null,
"description": "# PyWebTransport\n\n[](https://badge.fury.io/py/pywebtransport)\n[](https://pypi.org/project/pywebtransport/)\n[](https://opensource.org/licenses/MIT)\n[](https://github.com/lemonsterfy/pywebtransport/actions)\n[](https://codecov.io/gh/lemonsterfy/pywebtransport)\n[](https://docs.pywebtransport.org/en/latest/)\n\nAn async-native WebTransport stack for Python.\n\n## Features\n\n- **Full Async Support**: Built from the ground up on `asyncio` for high-performance, non-blocking I/O.\n- **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.\n- **Advanced Messaging**: Built-in managers for Pub/Sub and RPC (JSON-RPC 2.0 compliant), plus pluggable serializers (`JSON`, `MsgPack`, `Protobuf`) for structured data.\n- **Complete Protocol Implementation**: Full support for bidirectional and unidirectional streams, as well as unreliable datagrams.\n- **Lifecycle and Resource Management**: Robust, async context-managed components for handling connections, sessions, streams, monitoring, pooling, and resource management.\n- **Event-Driven Architecture**: A powerful `EventEmitter` system for decoupled, asynchronous communication between components.\n- **Type-Safe and Tested**: A fully type-annotated API with extensive test coverage (unit, integration, E2E) to ensure reliability and maintainability.\n\n## Installation\n\n```bash\npip install pywebtransport\n```\n\nFor more detailed instructions, including virtual environments and platform-specific notes, see the [Installation Guide](docs/installation.md).\n\n## Quick Start\n\n### Server\n\n```python\n# server.py\nimport asyncio\n\nfrom pywebtransport import (\n ConnectionError,\n ServerApp,\n ServerConfig,\n SessionError,\n WebTransportSession,\n WebTransportStream,\n)\nfrom pywebtransport.utils import generate_self_signed_cert\n\ngenerate_self_signed_cert(hostname=\"localhost\")\n\napp = ServerApp(\n config=ServerConfig(\n certfile=\"localhost.crt\",\n keyfile=\"localhost.key\",\n initial_max_data=1024 * 1024,\n initial_max_streams_bidi=10,\n )\n)\n\n\nasync def handle_datagrams(session: WebTransportSession) -> None:\n try:\n datagram_transport = await session.create_datagram_transport()\n while True:\n data = await datagram_transport.receive()\n await datagram_transport.send(data=b\"ECHO: \" + data)\n except (ConnectionError, SessionError, asyncio.CancelledError):\n pass\n\n\nasync def handle_streams(session: WebTransportSession) -> None:\n try:\n async for stream in session.incoming_streams():\n if isinstance(stream, WebTransportStream):\n data = await stream.read_all()\n await stream.write_all(data=b\"ECHO: \" + data)\n except (ConnectionError, SessionError, asyncio.CancelledError):\n pass\n\n\n@app.route(path=\"/\")\nasync def echo_handler(session: WebTransportSession) -> None:\n datagram_task = asyncio.create_task(handle_datagrams(session))\n stream_task = asyncio.create_task(handle_streams(session))\n try:\n await session.wait_closed()\n finally:\n datagram_task.cancel()\n stream_task.cancel()\n\n\nif __name__ == \"__main__\":\n app.run(host=\"127.0.0.1\", port=4433)\n```\n\n### Client\n\n```python\n# client.py\nimport asyncio\nimport ssl\n\nfrom pywebtransport import ClientConfig, WebTransportClient\n\n\nasync def main() -> None:\n config = ClientConfig(\n verify_mode=ssl.CERT_NONE,\n initial_max_data=1024 * 1024,\n initial_max_streams_bidi=10,\n )\n\n async with WebTransportClient(config=config) as client:\n session = await client.connect(url=\"https://127.0.0.1:4433/\")\n\n print(\"Connection established. Testing datagrams...\")\n datagram_transport = await session.create_datagram_transport()\n await datagram_transport.send(data=b\"Hello, Datagram!\")\n response = await datagram_transport.receive()\n print(f\"Datagram echo: {response!r}\\n\")\n\n print(\"Testing streams...\")\n stream = await session.create_bidirectional_stream()\n await stream.write_all(data=b\"Hello, Stream!\")\n response = await stream.read_all()\n print(f\"Stream echo: {response!r}\")\n\n await session.close()\n\n\nif __name__ == \"__main__\":\n try:\n asyncio.run(main())\n except KeyboardInterrupt:\n pass\n```\n\n## Documentation\n\n- **[Installation Guide](docs/installation.md)** - Learn how to install the library.\n- **[Quick Start](docs/quickstart.md)** - Discover how to get started with basic client and server setup.\n- **[API Reference](docs/api-reference/)** - Explore the complete API documentation.\n\n## Requirements\n\n- Python 3.11+\n- asyncio support\n- TLS 1.3\n\n**Dependencies:**\n\n- aioquic >= 1.3.0\n- cryptography >= 45.0.4\n\n## Contributing\n\nContributions are welcome! Please read our [Contributing Guide](CONTRIBUTING.md) for details on the development setup, testing, and pull request process.\n\n**Development Setup:**\n\n```bash\ngit clone https://github.com/lemonsterfy/pywebtransport.git\ncd pywebtransport\npip install -r dev-requirements.txt\npip install -e .\ntox\n```\n\n## License\n\nThis project is licensed under the MIT License - see the [LICENSE](LICENSE) file for details.\n\n## Acknowledgments\n\n- <a href=\"https://www.fastly.com/\" target=\"_blank\" rel=\"noopener noreferrer\">\n <img src=\"https://upload.wikimedia.org/wikipedia/commons/8/8a/Fastly_logo.svg\" alt=\"Fastly\" width=\"80\">\n </a>\n for providing critical infrastructure and services.\n- [aioquic](https://github.com/aiortc/aioquic) for the underlying QUIC protocol implementation.\n- [WebTransport Working Group](https://datatracker.ietf.org/wg/webtrans/) for defining and standardizing the WebTransport protocol.\n\n## Support\n\n- **Issues**: [GitHub Issues](https://github.com/lemonsterfy/pywebtransport/issues)\n- **Discussions**: [GitHub Discussions](https://github.com/lemonsterfy/pywebtransport/discussions)\n- **Email**: lemonsterfy@gmail.com\n",
"bugtrack_url": null,
"license": null,
"summary": "An async-native WebTransport stack for Python.",
"version": "0.8.1",
"project_urls": {
"Changelog": "https://github.com/lemonsterfy/pywebtransport/blob/main/CHANGELOG.md",
"Documentation": "https://docs.pywebtransport.org",
"Homepage": "https://github.com/lemonsterfy/pywebtransport",
"Issues": "https://github.com/lemonsterfy/pywebtransport/issues",
"Repository": "https://github.com/lemonsterfy/pywebtransport.git"
},
"split_keywords": [
"async",
" http3",
" networking",
" protocol",
" quic",
" webtransport"
],
"urls": [
{
"comment_text": null,
"digests": {
"blake2b_256": "150908912ed0e4f88566bb133221a3a84ad90e787c4238a9e2487402880ed1b2",
"md5": "b4298a00747e8f4e5637f99f3bbff921",
"sha256": "5efdb4a8655ee7a34f48445f52cbcbc1c475cff96e213adff5a10db8f29e5345"
},
"downloads": -1,
"filename": "pywebtransport-0.8.1-py3-none-any.whl",
"has_sig": false,
"md5_digest": "b4298a00747e8f4e5637f99f3bbff921",
"packagetype": "bdist_wheel",
"python_version": "py3",
"requires_python": ">=3.11",
"size": 140250,
"upload_time": "2025-11-04T04:38:37",
"upload_time_iso_8601": "2025-11-04T04:38:37.911481Z",
"url": "https://files.pythonhosted.org/packages/15/09/08912ed0e4f88566bb133221a3a84ad90e787c4238a9e2487402880ed1b2/pywebtransport-0.8.1-py3-none-any.whl",
"yanked": false,
"yanked_reason": null
},
{
"comment_text": null,
"digests": {
"blake2b_256": "d4ae1314a4a409f9b1ffa2963ef285f3507829581014ef9ecd61737bb32aaa99",
"md5": "c59d0ddbb9cb82f769abf896ca558b0e",
"sha256": "ad46c7a4a4c7004c665b94284a693f340ad3c1786a647bed390e48409b0aab77"
},
"downloads": -1,
"filename": "pywebtransport-0.8.1.tar.gz",
"has_sig": false,
"md5_digest": "c59d0ddbb9cb82f769abf896ca558b0e",
"packagetype": "sdist",
"python_version": "source",
"requires_python": ">=3.11",
"size": 113471,
"upload_time": "2025-11-04T04:38:36",
"upload_time_iso_8601": "2025-11-04T04:38:36.559644Z",
"url": "https://files.pythonhosted.org/packages/d4/ae/1314a4a409f9b1ffa2963ef285f3507829581014ef9ecd61737bb32aaa99/pywebtransport-0.8.1.tar.gz",
"yanked": false,
"yanked_reason": null
}
],
"upload_time": "2025-11-04 04:38:36",
"github": true,
"gitlab": false,
"bitbucket": false,
"codeberg": false,
"github_user": "lemonsterfy",
"github_project": "pywebtransport",
"travis_ci": false,
"coveralls": false,
"github_actions": true,
"lcname": "pywebtransport"
}
JSON
