### Set Up Python Virtual Environment

Source: https://github.com/wtransport/pywebtransport/blob/main/CONTRIBUTING.md

Commands to install the required Python version using pyenv and create/activate a virtual environment for development.

```bash
pyenv install
python -m venv .venv
source .venv/bin/activate
```

--------------------------------

### Install Project Dependencies

Source: https://github.com/wtransport/pywebtransport/blob/main/CONTRIBUTING.md

Installs development dependencies from 'dev-requirements.txt' and the project itself in editable mode.

```bash
pip install -r dev-requirements.txt
pip install -e .
```

--------------------------------

### ClientFleet - Load Distribution

Source: https://context7.com/wtransport/pywebtransport/llms.txt

This example demonstrates how to use ClientFleet to manage multiple WebTransportClient instances for load distribution. It covers concurrent connection establishment, round-robin client selection, and parallel communication.

```APIDOC
## ClientFleet - Load Distribution

### Description
Manages multiple `WebTransportClient` instances for load distribution using round-robin selection and concurrent connection establishment.

### Method
N/A (This is a client-side implementation example)

### Endpoint
N/A (Demonstrates client-side logic)

### Parameters
N/A

### Request Example
```python
import asyncio
import ssl
from pywebtransport import ClientConfig, WebTransportClient
from pywebtransport.client.fleet import ClientFleet

async def fleet_example() -> None:
    # Create multiple clients with different configurations
    configs = [
        ClientConfig(verify_mode=ssl.CERT_NONE, user_agent="Fleet-Worker-1"),
        ClientConfig(verify_mode=ssl.CERT_NONE, user_agent="Fleet-Worker-2"),
        ClientConfig(verify_mode=ssl.CERT_NONE, user_agent="Fleet-Worker-3"),
    ]

    clients = [WebTransportClient(config=cfg) for cfg in configs]

    fleet = ClientFleet(
        clients=clients,
        max_concurrent_handshakes=50,  # Limit concurrent connection attempts
    )

    async with fleet:
        print(f"Fleet size: {fleet.get_client_count()}")

        # Connect all clients to the same endpoint
        sessions = await fleet.connect_all(
            url="https://127.0.0.1:4433/",
            subprotocols=["binary"],
        )
        print(f"Connected {len(sessions)} sessions")

        # Use sessions for parallel work
        async def send_message(session, msg: str) -> None:
            stream = await session.create_bidirectional_stream()
            await stream.write_all(data=msg.encode(), end_stream=True)
            response = await stream.read_all()
            print(f"Session {session.session_id}: {response}")

        await asyncio.gather(
            *[send_message(s, f"Message {i}") for i, s in enumerate(sessions)]
        )

        # Round-robin client selection for new connections
        for i in range(10):
            client = fleet.get_client()
            session = await client.connect(url="https://127.0.0.1:4433/")
            print(f"Request {i} using client index: {fleet._current_index}")
            await session.close()

        # Close all sessions
        for session in sessions:
            await session.close()

if __name__ == "__main__":
    asyncio.run(fleet_example())
```

### Response
N/A (Client-side output)

### Response Example
```
Fleet size: 3
Connected 3 sessions
Session <session_id_1>: b'Message 0'
Session <session_id_2>: b'Message 1'
Session <session_id_3>: b'Message 2'
Request 0 using client index: 1
Request 1 using client index: 2
Request 2 using client index: 0
Request 3 using client index: 1
Request 4 using client index: 2
Request 5 using client index: 0
Request 6 using client index: 1
Request 7 using client index: 2
Request 8 using client index: 0
Request 9 using client index: 1
```
```

--------------------------------

### ReconnectingClient - Auto-Reconnection with Backoff

Source: https://context7.com/wtransport/pywebtransport/llms.txt

This example shows how to use the ReconnectingClient to automatically re-establish connections with exponential backoff. It also demonstrates setting up event handlers for connection status changes and using the established session for communication.

```APIDOC
## ReconnectingClient - Auto-Reconnection with Backoff

### Description
Provides automatic reconnection with exponential backoff, connection state events, and session retrieval.

### Method
N/A (This is a client-side implementation example)

### Endpoint
N/A (Demonstrates client-side logic)

### Parameters
N/A

### Request Example
```python
import asyncio
import ssl
from pywebtransport import ClientConfig, WebTransportClient
from pywebtransport.client.reconnecting import ReconnectingClient
from pywebtransport.types import EventType

async def resilient_client() -> None:
    config = ClientConfig(
        verify_mode=ssl.CERT_NONE,
        connect_timeout=10.0,
        max_connection_retries=10,  # Max retry attempts
        retry_delay=1.0,            # Initial delay between retries
        retry_backoff=2.0,          # Exponential backoff multiplier
        max_retry_delay=30.0,       # Maximum delay cap
    )

    async with WebTransportClient(config=config) as client:
        reconnecting = ReconnectingClient(
            url="https://127.0.0.1:4433/",
            client=client,
            subprotocols=["chat"],
        )

        # Set up connection event handlers
        async def on_connected(event) -> None:
            print(f"Connected! Attempt: {event.data.get('attempt')}")
            session = event.data.get("session")
            if session:
                print(f"Session ID: {session.session_id}")

        async def on_lost(event) -> None:
            print(f"Connection lost to {event.data.get('url')}")

        async def on_failed(event) -> None:
            print(f"Connection failed: {event.data.get('reason')}")
            print(f"Last error: {event.data.get('last_error')}")

        reconnecting.on(event_type=EventType.CONNECTION_ESTABLISHED, handler=on_connected)
        reconnecting.on(event_type=EventType.CONNECTION_LOST, handler=on_lost)
        reconnecting.on(event_type=EventType.CONNECTION_FAILED, handler=on_failed)

        async with reconnecting:
            # Wait for initial connection
            session = await reconnecting.get_session(wait_timeout=30.0)
            print(f"Got session: {session.session_id}")

            # Use the session
            stream = await session.create_bidirectional_stream()
            await stream.write_all(data=b"Hello", end_stream=True)
            response = await stream.read_all()
            print(f"Response: {response}")

            # Check connection status
            if reconnecting.is_connected:
                print("Currently connected")

            # Get session again (automatically reconnects if needed)
            session = await reconnecting.get_session()
            await session.send_datagram(data=b"Quick message")

            # Keep running to demonstrate reconnection
            await asyncio.sleep(60)

if __name__ == "__main__":
    asyncio.run(resilient_client())
```

### Response
N/A (Client-side output)

### Response Example
```
Connected! Attempt: 1
Session ID: <session_id>
Got session: <session_id>
Response: b'<response_data>'
Currently connected
```
```

--------------------------------

### Perform Certificate Generation and Header Manipulation

Source: https://context7.com/wtransport/pywebtransport/llms.txt

Provides examples for generating self-signed certificates for development and performing common HTTP header operations like case-insensitive lookups and merging.

```python
from pywebtransport.utils import (
    generate_self_signed_cert,
    find_header,
    find_header_str,
    merge_headers,
    ensure_buffer,
    format_duration,
)

generate_self_signed_cert(hostname="localhost")
generate_self_signed_cert(hostname="example.com")

headers_dict = {"content-type": "application/json", "x-custom": "value"}

content_type = find_header(headers=headers_dict, key="Content-Type")
print(f"Content-Type: {content_type}")

auth = find_header_str(headers=headers_dict, key="authorization", default="none")
print(f"Authorization: {auth}")

base = {"accept": "application/json", "x-version": "1"}
update = {"x-version": "2", "x-new": "value"}
merged = merge_headers(base=base, update=update)
print(f"Merged: {merged}")
```

--------------------------------

### WebTransportSession Management and Stream Handling

Source: https://context7.com/wtransport/pywebtransport/llms.txt

Demonstrates comprehensive management of a WebTransport session. It covers accessing session properties, retrieving diagnostics, creating and interacting with bidirectional and unidirectional streams, sending and receiving datagrams, and handling session events. It also includes examples of exporting TLS keying material and managing flow control credits for data and streams.

```python
import asyncio
from pywebtransport import WebTransportSession, WebTransportStream
from pywebtransport.types import EventType

async def handle_session(session: WebTransportSession) -> None:
    async with session:
        # Access session properties
        print(f"Session ID: {session.session_id}")
        print(f"Path: {session.path}")
        print(f"Remote address: {session.remote_address}")
        print(f"Headers: {session.headers}")
        print(f"Negotiated subprotocol: {session.subprotocol}")
        print(f"State: {session.state}")

        # Get session diagnostics
        diag = await session.diagnostics()
        print(f"Active streams: {diag.active_streams}")
        print(f"Datagrams sent: {diag.datagrams_sent}")
        print(f"Local max data: {diag.local_max_data}")

        # Create and use streams
        bidi_stream = await session.create_bidirectional_stream()
        await bidi_stream.write_all(data=b"Hello, bidirectional!", end_stream=True)
        response = await bidi_stream.read_all()
        print(f"Received: {response}")

        uni_stream = await session.create_unidirectional_stream()
        await uni_stream.write_all(data=b"One-way message", end_stream=True)

        # Send unreliable datagrams
        await session.send_datagram(data=b"Quick unreliable message")

        # Listen for datagram events
        async def on_datagram(event) -> None:
            if isinstance(event.data, dict) and (data := event.data.get("data")):
                print(f"Datagram received: {data!r}")

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

        # Accept incoming streams using async generators
        async for stream in session.incoming_bidirectional_streams():
            asyncio.create_task(process_stream(stream))

        # Or accept individual streams
        try:
            uni_recv = await session.accept_unidirectional_stream()
            data = await uni_recv.read_all()
            print(f"Received from unidirectional: {data}")
        except Exception as e:
            print(f"Error: {e}")

        # Export TLS keying material (for E2E encryption)
        keying_material = await session.export_keying_material(
            label="EXPORTER-my-app",
            context=b"session-context",
            length=32,
        )
        print(f"Exported key: {keying_material.hex()}")

        # Flow control management
        await session.grant_data_credit(max_data=1048576)  # 1MB
        await session.grant_streams_credit(is_unidirectional=False, max_streams=100)

async def process_stream(stream: WebTransportStream) -> None:
    async with stream:
        data = await stream.read_all()
        await stream.write_all(data=b"Processed: " + data, end_stream=True)
```

--------------------------------

### Implement ServerApp with Routing

Source: https://context7.com/wtransport/pywebtransport/llms.txt

Shows how to initialize a ServerApp, configure server settings, and define route handlers for incoming WebTransport streams.

```python
import asyncio
from pywebtransport import ServerApp, ServerConfig, WebTransportSession, WebTransportStream
from pywebtransport.utils import generate_self_signed_cert

generate_self_signed_cert(hostname="localhost")

config = ServerConfig(
    certfile="localhost.crt",
    keyfile="localhost.key",
    bind_host="0.0.0.0",
    bind_port=4433,
    max_connections=1000,
    max_sessions=10000,
    connection_idle_timeout=300.0,
)

app = ServerApp(config=config)

@app.on_startup
async def startup_handler() -> None:
    print("Server starting up...")

@app.on_shutdown
async def shutdown_handler() -> None:
    print("Server shutting down...")

@app.route(path="/echo")
async def echo_handler(session: WebTransportSession) -> None:
    async for stream in session.incoming_bidirectional_streams():
        asyncio.create_task(handle_echo_stream(stream))

async def handle_echo_stream(stream: WebTransportStream) -> None:
    async with stream:
        data = await stream.read_all()
        await stream.write_all(data=b"ECHO: " + data, end_stream=True)
```

--------------------------------

### Install PyWebTransport via pip

Source: https://github.com/wtransport/pywebtransport/blob/main/README.md

Command to install the PyWebTransport package using the Python package manager.

```bash
pip install pywebtransport
```

--------------------------------

### Create a WebTransport Server

Source: https://github.com/wtransport/pywebtransport/blob/main/README.md

Demonstrates how to initialize a ServerApp, generate self-signed certificates, and handle incoming datagrams and bidirectional streams.

```python
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"))


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

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

    try:
        async for stream in session.incoming_bidirectional_streams():
            asyncio.create_task(handle_stream(stream))
    finally:
        session.events.off(event_type=EventType.DATAGRAM_RECEIVED, handler=on_datagram)


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


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

--------------------------------

### Create a WebTransport Client

Source: https://github.com/wtransport/pywebtransport/blob/main/README.md

Shows how to configure a client, connect to a server, send datagrams, and perform bidirectional stream communication.

```python
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)

    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)
        if isinstance(event.data, dict) and (data := event.data.get("data")):
            print(f"Datagram: {data!r}")

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

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

        await session.close()


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

--------------------------------

### Sign Commits for PyWebTransport

Source: https://github.com/wtransport/pywebtransport/blob/main/CONTRIBUTING.md

Example of how to sign off a Git commit, which is mandatory for all contributions to PyWebTransport, adhering to the Developer Certificate of Origin (DCO).

```bash
git commit -s
```

--------------------------------

### Configuration Manipulation

Source: https://context7.com/wtransport/pywebtransport/llms.txt

Demonstrates updating, converting to dictionary, creating from dictionary, and copying configuration objects.

```APIDOC
## Configuration Manipulation

### Description
PyWebTransport configuration objects (`ClientConfig`, `ServerConfig`) support dynamic updates, serialization to dictionaries, creation from dictionaries, and copying.

### Methods
- **update(**kwargs**)**: Updates configuration with provided keyword arguments. Returns a new configuration object with the updates.
- **to_dict()**: Serializes the configuration object into a dictionary.
- **from_dict(config_dict)**: Class method to create a configuration object from a dictionary.
- **copy()**: Creates a deep copy of the configuration object.

### Request Example
```python
from pywebtransport import ClientConfig

# Initial configuration
client_config = ClientConfig(connect_timeout=10.0, max_connections=100)

# Update configuration (returns a new object)
updated_config = client_config.update(connect_timeout=20.0, max_connections=200)

# Convert to dictionary
config_dict = client_config.to_dict()

# Create from dictionary
from_dict_config = ClientConfig.from_dict(config_dict=config_dict)

# Copy configuration
copied_config = client_config.copy()

# Accessing configuration values
print(f"Connect timeout: {client_config.connect_timeout}")
print(f"Verify mode: {client_config.verify_mode.name if client_config.verify_mode else 'Not Set'}")
```

### Response
N/A (Demonstration of object manipulation)

### Response Example
N/A
```

--------------------------------

### Implement Event-Driven Architecture with EventEmitter

Source: https://context7.com/wtransport/pywebtransport/llms.txt

Demonstrates how to use the EventEmitter class to manage event subscriptions, emit events, handle queues, and manage history. It covers both persistent and one-time handlers, as well as waiting for specific events with timeouts.

```python
import asyncio
from pywebtransport.events import Event, EventEmitter
from pywebtransport.types import EventType

async def event_system_demo() -> None:
    emitter = EventEmitter(
        max_listeners=100,
        max_history=1000,
        max_queue_size=500,
    )

    async def on_data_received(event: Event) -> None:
        print(f"Data: {event.data}")
        print(f"Timestamp: {event.timestamp}")
        print(f"Source: {event.source}")

    emitter.on(event_type=EventType.DATAGRAM_RECEIVED, handler=on_data_received)

    def on_first_connection(event: Event) -> None:
        print(f"First connection established!")

    emitter.once(event_type=EventType.CONNECTION_ESTABLISHED, handler=on_first_connection)

    async def log_all_events(event: Event) -> None:
        print(f"[LOG] {event.type}: {event.data}")

    emitter.on_any(handler=log_all_events)

    await emitter.emit(
        event_type=EventType.DATAGRAM_RECEIVED,
        data={"data": b"Hello", "size": 5},
        source="client-123",
    )

    emitter.emit_nowait(
        event_type=EventType.CONNECTION_ESTABLISHED,
        data={"session_id": 12345},
    )

    async def wait_for_specific_event() -> None:
        try:
            event = await emitter.wait_for(
                event_type=EventType.SESSION_CLOSED,
                timeout=5.0,
                condition=lambda e: e.data and e.data.get("clean") is True,
            )
            print(f"Session closed cleanly: {event}")
        except asyncio.TimeoutError:
            print("Timeout waiting for clean session close")

    async def wait_for_any_close() -> None:
        event = await emitter.wait_for(
            event_type=[EventType.CONNECTION_CLOSED, EventType.SESSION_CLOSED],
            timeout=10.0,
        )
        print(f"Close event received: {event.type}")

    emitter.pause()
    emitter.emit_nowait(event_type=EventType.STREAM_OPENED, data={"stream_id": 1})
    emitter.emit_nowait(event_type=EventType.STREAM_OPENED, data={"stream_id": 2})
    task = emitter.resume()
    if task:
        await task

    history = emitter.get_event_history(event_type=EventType.DATAGRAM_RECEIVED, limit=10)
    for event in history:
        print(f"Historical: {event}")

    stats = emitter.get_stats()
    print(f"Total handlers: {stats['total_handlers']}")
    print(f"Event types: {stats['event_types']}")
    print(f"History size: {stats['history_size']}")
    print(f"Queued events: {stats['queued_events']}")

    emitter.off(event_type=EventType.DATAGRAM_RECEIVED, handler=on_data_received)
    emitter.off_any(handler=log_all_events)
    emitter.remove_all_listeners()
    emitter.clear_history()
    await emitter.close()

if __name__ == "__main__":
    asyncio.run(event_system_demo())
```

--------------------------------

### Server Configuration with PyWebTransport

Source: https://context7.com/wtransport/pywebtransport/llms.txt

Demonstrates the creation of a server configuration object for PyWebTransport. It includes essential TLS settings (certificate and key), optional client certificate verification (mTLS), bind addresses and ports, server-side limits, and access to all base configuration options.

```python
import ssl
from pywebtransport import ServerConfig

server_config = ServerConfig(
    certfile="/path/to/server.crt",
    keyfile="/path/to/server.key",

    ca_certs="/path/to/client-ca.crt",
    verify_mode=ssl.CERT_OPTIONAL,  # or CERT_REQUIRED for mandatory mTLS

    bind_host="0.0.0.0",
    bind_port=4433,

    max_connections=10000,
    max_sessions=100000,
    connection_idle_timeout=300.0,

    congestion_control_algorithm="cubic",
    initial_max_data=10485760,
    initial_max_streams_bidi=1000,
    initial_max_streams_uni=1000,
)
```

--------------------------------

### ServerConfig

Source: https://context7.com/wtransport/pywebtransport/llms.txt

Detailed configuration options for setting up a WebTransport server.

```APIDOC
## Server Configuration - ServerConfig

### Description
The `ServerConfig` class provides comprehensive settings for a WebTransport server, including essential TLS configurations, bind addresses, server limits, and all base configuration options applicable to clients.

### Method
`ServerConfig(...)`

### Endpoint
N/A (Server-side configuration)

### Parameters

#### Required TLS Settings
- **certfile** (str) - Required - Path to the server's certificate file.
- **keyfile** (str) - Required - Path to the server's private key file.

#### Optional TLS Settings (for mTLS)
- **ca_certs** (str) - Optional - Path to CA certificate file for verifying client certificates.
- **verify_mode** (int) - Optional - SSL verification mode for client certificates (e.g., `ssl.CERT_OPTIONAL`, `ssl.CERT_REQUIRED`).

#### Bind Settings
- **bind_host** (str) - Optional - Host address to bind the server to (e.g., `"0.0.0.0"`).
- **bind_port** (int) - Optional - Port number to bind the server to (e.g., `4433`).

#### Server Limits
- **max_connections** (int) - Optional - Maximum number of concurrent connections allowed.
- **max_sessions** (int) - Optional - Maximum number of concurrent sessions allowed.
- **connection_idle_timeout** (float) - Optional - Timeout in seconds for an idle connection before it's closed.

#### Inherited BaseConfig Options
All parameters available in `ClientConfig` under Protocol Settings, Flow Control, Buffer and Message Limits, Event System, and Timeouts are also applicable here.

### Request Example
```python
import ssl
from pywebtransport import ServerConfig

server_config = ServerConfig(
    certfile="/path/to/server.crt",
    keyfile="/path/to/server.key",
    ca_certs="/path/to/client-ca.crt",
    verify_mode=ssl.CERT_OPTIONAL,
    bind_port=4433
)
```

### Response
N/A (Configuration object)

### Response Example
N/A
```

--------------------------------

### ClientConfig

Source: https://context7.com/wtransport/pywebtransport/llms.txt

Detailed configuration options for setting up a WebTransport client.

```APIDOC
## Client Configuration - ClientConfig

### Description
The `ClientConfig` class allows for detailed configuration of a WebTransport client, covering TLS settings, connection parameters, retry logic, protocol options, flow control, buffer limits, event system, timeouts, and HTTP headers.

### Method
`ClientConfig(...)`

### Endpoint
N/A (Client-side configuration)

### Parameters

#### TLS Settings
- **verify_mode** (int) - Optional - SSL verification mode (e.g., `ssl.CERT_REQUIRED`).
- **ca_certs** (str) - Optional - Path to CA certificate bundle for verifying server identity.
- **certfile** (str) - Optional - Path to client certificate file for mutual TLS (mTLS).
- **keyfile** (str) - Optional - Path to client private key file for mTLS.

#### Connection Settings
- **connect_timeout** (float) - Optional - Timeout in seconds for establishing a connection.
- **connection_attempt_delay** (float) - Optional - Delay in seconds for subsequent connection attempts (Happy Eyeballs).
- **connection_idle_timeout** (float) - Optional - Timeout in seconds for an idle connection before it's closed.
- **max_connections** (int) - Optional - Maximum number of concurrent connections.
- **max_sessions** (int) - Optional - Maximum number of concurrent sessions.

#### Retry Settings
- **max_connection_retries** (int) - Optional - Maximum number of retries for connection attempts.
- **retry_delay** (float) - Optional - Initial delay in seconds between retries.
- **retry_backoff** (float) - Optional - Multiplier for increasing retry delay.
- **max_retry_delay** (float) - Optional - Maximum delay in seconds between retries.

#### Protocol Settings
- **alpn_protocols** (list[str]) - Optional - Application-Layer Protocol Negotiation list (e.g., `["h3"]`).
- **congestion_control_algorithm** (str) - Optional - Congestion control algorithm (e.g., `"cubic"`, `"reno"`, `"bbr"`).
- **keep_alive** (float) - Optional - Keep-alive interval in seconds.
- **subprotocols** (list[str]) - Optional - List of supported subprotocols (e.g., `["chat", "binary"]`).

#### Flow Control
- **initial_max_data** (int) - Optional - Initial maximum data flow control window size.
- **initial_max_streams_bidi** (int) - Optional - Initial maximum bidirectional stream count.
- **initial_max_streams_uni** (int) - Optional - Initial maximum unidirectional stream count.
- **flow_control_window_size** (int) - Optional - Default flow control window size per stream.
- **flow_control_window_auto_scale** (bool) - Optional - Whether to auto-scale flow control windows.

#### Buffer and Message Limits
- **max_datagram_size** (int) - Optional - Maximum size of a datagram.
- **max_message_size** (int) - Optional - Maximum size of a reliable message.
- **max_stream_read_buffer** (int) - Optional - Maximum read buffer size per stream.
- **max_stream_write_buffer** (int) - Optional - Maximum write buffer size per stream.
- **max_capsule_size** (int) - Optional - Maximum size of a capsule.

#### Event System
- **max_event_listeners** (int) - Optional - Maximum number of event listeners.
- **max_event_queue_size** (int) - Optional - Maximum size of the event queue.
- **max_event_history_size** (int) - Optional - Maximum size of event history.

#### Timeouts
- **read_timeout** (float) - Optional - Timeout for read operations.
- **write_timeout** (float) - Optional - Timeout for write operations.
- **stream_creation_timeout** (float) - Optional - Timeout for stream creation.
- **close_timeout** (float) - Optional - Timeout for connection closure.

#### HTTP Settings
- **headers** (dict) - Optional - Custom HTTP headers to send with the connection request.
- **user_agent** (str) - Optional - User-Agent string for the client.

#### Logging
- **log_level** (str) - Optional - Logging level (e.g., `"INFO"`, `"DEBUG"`).

### Request Example
```python
import ssl
from pywebtransport import ClientConfig

client_config = ClientConfig(
    verify_mode=ssl.CERT_REQUIRED,
    ca_certs="/path/to/ca-bundle.crt",
    connect_timeout=10.0,
    headers={"x-client-id": "my-app"},
    log_level="INFO"
)
client_config.validate()
```

### Response
N/A (Configuration object)

### Response Example
N/A
```

--------------------------------

### Configure Server Middleware for WebTransport

Source: https://context7.com/wtransport/pywebtransport/llms.txt

Illustrates how to integrate and customize server middleware for authentication, rate limiting, and CORS validation. It shows how to use built-in factories and implement custom middleware logic via decorators.

```python
from pywebtransport import ServerApp, ServerConfig
from pywebtransport.server.middleware import (
    create_auth_middleware,
    create_cors_middleware,
    create_logging_middleware,
    create_rate_limit_middleware,
    MiddlewareRejected
)

config = ServerConfig(certfile="cert.pem", keyfile="key.pem")
app = ServerApp(config=config)

app.add_middleware(middleware=create_logging_middleware())
app.add_middleware(middleware=create_cors_middleware(allowed_origins=["https://*.example.com"]))
app.add_middleware(middleware=create_rate_limit_middleware(max_requests=100, window_seconds=60))

async def verify_token(*, headers) -> bool:
    return headers.get("authorization") == "Bearer valid_token_123"

app.add_middleware(middleware=create_auth_middleware(auth_handler=verify_token))

@app.middleware
async def custom_middleware(*, session) -> None:
    if "Bot" in str(session.headers.get("user-agent", "")):
        raise MiddlewareRejected(status_code=403)
```

--------------------------------

### Server Middleware Configuration

Source: https://context7.com/wtransport/pywebtransport/llms.txt

Documentation for configuring server middleware including authentication, rate limiting, and CORS policies.

```APIDOC
## Server Middleware Configuration

### Description
The middleware system allows for request processing chains. It provides built-in handlers for common security and traffic management tasks.

### Method
N/A (Configuration-based)

### Parameters
#### Middleware Types
- **create_auth_middleware** - Required - Accepts an auth_handler function to validate headers.
- **create_rate_limit_middleware** - Required - Configures max_requests and window_seconds for traffic control.
- **create_cors_middleware** - Required - Sets allowed_origins for cross-origin requests.

### Request Example
app.add_middleware(middleware=create_rate_limit_middleware(max_requests=100, window_seconds=60))

### Response
#### Success Response
- **MiddlewareRejected** (Exception) - Raised when a request fails middleware validation, returning a 403 status code.
```

--------------------------------

### Manage WebTransport Connections with WebTransportClient

Source: https://context7.com/wtransport/pywebtransport/llms.txt

Demonstrates how to configure and use the WebTransportClient to manage connections, sessions, and diagnostics. It includes setting custom headers, subprotocols, and retry logic.

```python
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,
        connect_timeout=10.0,
        max_connections=100,
        max_sessions=1000,
        user_agent="MyApp/1.0",
        headers={"x-custom-header": "value"},
        subprotocols=["chat", "binary"],
        max_connection_retries=5,
        retry_delay=1.0,
        retry_backoff=2.0,
        max_retry_delay=30.0,
    )

    async with WebTransportClient(config=config) as client:
        client.set_default_headers(headers={"authorization": "Bearer token123"})
        session = await client.connect(
            url="https://127.0.0.1:4433/echo",
            timeout=5.0,
            headers={"x-request-id": "abc123"},
            subprotocols=["chat"],
        )
        diagnostics = await client.diagnostics()
        print(f"Connections attempted: {diagnostics.stats.connections_attempted}")
        print(f"Success rate: {diagnostics.stats.success_rate:.2%}")
        print(f"Avg connect time: {diagnostics.stats.avg_connect_time:.3f}s")
        print(f"Connection states: {diagnostics.connection_states}")
        print(f"Health issues: {diagnostics.issues}")
        await session.close()

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

--------------------------------

### Manage WebTransport Load Distribution with ClientFleet

Source: https://context7.com/wtransport/pywebtransport/llms.txt

Shows how to utilize ClientFleet to manage multiple WebTransportClient instances. It demonstrates concurrent connection establishment, parallel message processing, and round-robin client selection.

```python
import asyncio
import ssl
from pywebtransport import ClientConfig, WebTransportClient
from pywebtransport.client.fleet import ClientFleet

async def fleet_example() -> None:
    configs = [
        ClientConfig(verify_mode=ssl.CERT_NONE, user_agent="Fleet-Worker-1"),
        ClientConfig(verify_mode=ssl.CERT_NONE, user_agent="Fleet-Worker-2"),
        ClientConfig(verify_mode=ssl.CERT_NONE, user_agent="Fleet-Worker-3"),
    ]

    clients = [WebTransportClient(config=cfg) for cfg in configs]

    fleet = ClientFleet(
        clients=clients,
        max_concurrent_handshakes=50,
    )

    async with fleet:
        print(f"Fleet size: {fleet.get_client_count()}")
        sessions = await fleet.connect_all(
            url="https://127.0.0.1:4433/",
            subprotocols=["binary"],
        )
        print(f"Connected {len(sessions)} sessions")

        async def send_message(session, msg: str) -> None:
            stream = await session.create_bidirectional_stream()
            await stream.write_all(data=msg.encode(), end_stream=True)
            response = await stream.read_all()
            print(f"Session {session.session_id}: {response}")

        await asyncio.gather(
            *[send_message(s, f"Message {i}") for i, s in enumerate(sessions)]
        )

        for i in range(10):
            client = fleet.get_client()
            session = await client.connect(url="https://127.0.0.1:4433/")
            print(f"Request {i} using client index: {fleet._current_index}")
            await session.close()

        for session in sessions:
            await session.close()

if __name__ == "__main__":
    asyncio.run(fleet_example())
```

--------------------------------

### Client Configuration with PyWebTransport

Source: https://context7.com/wtransport/pywebtransport/llms.txt

Illustrates the creation and validation of a client configuration object for PyWebTransport. It covers TLS settings, connection parameters, retry logic, protocol options, flow control, buffer limits, event system configuration, timeouts, HTTP settings, and logging.

```python
import ssl
from pywebtransport import ClientConfig

client_config = ClientConfig(
    verify_mode=ssl.CERT_REQUIRED,
    ca_certs="/path/to/ca-bundle.crt",
    certfile="/path/to/client.crt",  # For mTLS
    keyfile="/path/to/client.key",   # For mTLS

    connect_timeout=10.0,
    connection_attempt_delay=0.25,  # Happy Eyeballs delay
    connection_idle_timeout=300.0,
    max_connections=100,
    max_sessions=1000,

    max_connection_retries=5,
    retry_delay=1.0,
    retry_backoff=2.0,
    max_retry_delay=30.0,

    alpn_protocols=["h3"],
    congestion_control_algorithm="cubic",  # or "reno", "bbr"
    keep_alive=30.0,
    subprotocols=["chat", "binary"],

    initial_max_data=1048576,
    initial_max_streams_bidi=100,
    initial_max_streams_uni=100,
    flow_control_window_size=65536,
    flow_control_window_auto_scale=True,

    max_datagram_size=65535,
    max_message_size=10485760,
    max_stream_read_buffer=262144,
    max_stream_write_buffer=262144,
    max_capsule_size=65536,

    max_event_listeners=100,
    max_event_queue_size=1000,
    max_event_history_size=100,

    read_timeout=30.0,
    write_timeout=30.0,
    stream_creation_timeout=10.0,
    close_timeout=5.0,

    headers={"x-client-id": "my-app"},
    user_agent="MyApp/1.0",

    log_level="INFO",
)

client_config.validate()

updated = client_config.update(connect_timeout=20.0, max_connections=200)
config_dict = client_config.to_dict()
from_dict = ClientConfig.from_dict(config_dict=config_dict)
copied = client_config.copy()

print(f"Connect timeout: {client_config.connect_timeout}")
print(f"Verify mode: {client_config.verify_mode.name}")
```

--------------------------------

### Utility Functions

Source: https://context7.com/wtransport/pywebtransport/llms.txt

Demonstrates the usage of utility functions for header merging, buffer validation, and duration formatting.

```APIDOC
## Utility Functions

### Description
This section covers utility functions provided by PyWebTransport for common tasks.

### Header Merging
Combines multiple lists of headers, with later lists overriding earlier ones.

```python
from pywebtransport.utils import merge_headers

base_list = [("accept", "application/json")]
update_list = [("x-custom", "value")]
merged_list = merge_headers(base=base_list, update=update_list)
print(f"Merged list: {merged_list}")
```

### Buffer Validation
Ensures data is in a buffer format (bytes), converting strings if necessary.

```python
from pywebtransport.utils import ensure_buffer

# Converts string to bytes
buffer_from_string = ensure_buffer(data="Hello, world!")

# Returns as-is if already bytes
buffer_from_bytes = ensure_buffer(data=b"Already bytes")

# Converts bytearray with specified encoding
buffer_from_bytearray = ensure_buffer(data=bytearray(b"Mutable"), encoding="utf-8")
```

### Duration Formatting
Formats a duration in seconds into a human-readable string with appropriate units (µs, ms, s, m, h).

```python
from pywebtransport.utils import format_duration

print(format_duration(seconds=0.000001))   # Expected output: "1.0µs"
print(format_duration(seconds=0.001))      # Expected output: "1.0ms"
print(format_duration(seconds=1.5))        # Expected output: "1.5s"
print(format_duration(seconds=90.5))       # Expected output: "1m30.5s"
print(format_duration(seconds=3661.5))     # Expected output: "1h1m1.5s"
```
```

--------------------------------

### Utility Functions: Header Merging, Buffer Validation, and Duration Formatting

Source: https://context7.com/wtransport/pywebtransport/llms.txt

Demonstrates the usage of utility functions for merging headers, ensuring data is in a buffer format (converting strings to bytes), and formatting duration values into human-readable strings with appropriate units (e.g., µs, ms, s, m, h).

```python
base_list = [("accept", "application/json")]
update_list = [("x-custom", "value")]
merged_list = merge_headers(base=base_list, update=update_list)
print(f"Merged list: {merged_list}")

buffer = ensure_buffer(data="Hello, world!")  # Converts string to bytes
buffer = ensure_buffer(data=b"Already bytes")  # Returns as-is
buffer = ensure_buffer(data=bytearray(b"Mutable"), encoding="utf-8")

print(format_duration(seconds=0.000001))   # "1.0µs"
print(format_duration(seconds=0.001))      # "1.0ms"
print(format_duration(seconds=1.5))        # "1.5s"
print(format_duration(seconds=90.5))       # "1m30.5s"
print(format_duration(seconds=3661.5))     # "1h1m1.5s"
```

--------------------------------

### Clone PyWebTransport Repository

Source: https://github.com/wtransport/pywebtransport/blob/main/CONTRIBUTING.md

Steps to fork the PyWebTransport repository, clone it locally, and set up the upstream remote for future updates.

```bash
git clone https://github.com/<your-username>/pywebtransport.git
cd pywebtransport
git remote add upstream https://github.com/wtransport/pywebtransport.git
```