### Complete WebSocket Session Example

Source: https://github.com/frankie567/httpx-ws/blob/main/_autodocs/api-reference/websocket-session.md

Demonstrates establishing a WebSocket connection, sending JSON data, and receiving messages with a timeout. Handles WebSocket disconnections gracefully. Ensure you have httpx and httpx-ws installed.

```python
import httpx
from httpx_ws import WebSocketClient

with httpx.Client() as client:
    ws_client = WebSocketClient(client)
    with ws_client.connect("ws://localhost:8000/chat") as ws:
        # Send a message
        ws.send_json({"type": "join", "username": "alice"})
        
        # Receive messages with timeout
        try:
            while True:
                message = ws.receive_json(timeout=10.0)
                if message["type"] == "message":
                    print(f"{message['user']}: {message['text']}")
        except WebSocketDisconnect as e:
            print(f"Disconnected with code {e.code}: {e.reason}")
```

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

### Complete Async WebSocket Client Example

Source: https://github.com/frankie567/httpx-ws/blob/main/_autodocs/api-reference/async-websocket-session.md

A full example demonstrating how to connect to a WebSocket, send a join message, and continuously receive and print chat messages until the connection is disconnected. It includes exception handling for WebSocketDisconnect.

```python
import httpx
from httpx_ws import AsyncWebSocketClient

async def main():
    async with httpx.AsyncClient() as client:
        ws_client = AsyncWebSocketClient(client)
        async with ws_client.connect("ws://localhost:8000/chat") as ws:
            # Send initial message
            await ws.send_json({"type": "join", "username": "alice"})
            
            # Receive messages
            try:
                while True:
                    message = await ws.receive_json(timeout=30.0)
                    if message["type"] == "message":
                        print(f"{message['user']}: {message['text']}")
            except WebSocketDisconnect as e:
                print(f"Disconnected: {e.reason}")
```

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

### ASGIWebSocketTransport Async Context Manager Example

Source: https://github.com/frankie567/httpx-ws/blob/main/_autodocs/api-reference/asgi-websocket-transport.md

Demonstrates how to use ASGIWebSocketTransport as an async context manager to test WebSocket connections with an ASGI application.

```python
async with ASGIWebSocketTransport(app) as transport:
    async with httpx.AsyncClient(transport=transport) as client:
        async with aconnect_ws("ws://localhost/ws", client=client) as ws:
            await ws.send_text("Hello")
            response = await ws.receive_text()
```

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

### Starlette ASGI App Example

Source: https://github.com/frankie567/httpx-ws/blob/main/docs/usage/asgi.md

Defines a basic Starlette application with both HTTP and WebSocket routes. This serves as the target application for testing.

```python
from starlette.applications import Starlette
from starlette.responses import HTMLResponse
from starlette.routing import Route, WebSocketRoute


async def http_hello(request):
    return HTMLResponse("Hello World!")

async def ws_hello(websocket):
    await websocket.accept()
    await websocket.send_text("Hello World!")
    await websocket.close()


app = Starlette(
    routes=[
        Route("/http", http_hello),
        WebSocketRoute("/ws", ws_hello),
    ],
)
```

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

### Connect to Production WebSocket API

Source: https://github.com/frankie567/httpx-ws/blob/main/_autodocs/configuration.md

This example shows how to connect to a production WebSocket API with authentication, a longer timeout, and configured keepalive settings to detect dead connections.

```python
with connect_ws(
    "wss://api.example.com/ws",
    auth=("api_key", ""),
    timeout=30.0,
    keepalive_ping_interval_seconds=20.0,
    keepalive_ping_timeout_seconds=10.0,  # Detect dead connection quickly
) as ws:
    ws.send_json({"command": "subscribe"})
    while True:
        data = ws.receive_json(timeout=60.0)
        process(data)
```

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

### Complete Chat Application Example

Source: https://github.com/frankie567/httpx-ws/blob/main/_autodocs/api-reference/websocket-client.md

This snippet shows how to establish a WebSocket connection, send JSON messages (like joining, leaving), and receive/process incoming JSON messages in a chat application. It includes error handling for timeouts and other exceptions, and demonstrates keeping the connection alive with pings.

```python
import httpx
from httpx_ws import WebSocketClient

def main():
    with httpx.Client() as client:
        ws_client = WebSocketClient(client)
        with ws_client.connect("ws://localhost:8000/chat") as ws:
            # Join the chat
            ws.send_json({
                "type": "join",
                "username": "alice"
            })
            
            # Handle incoming messages
            while True:
                try:
                    message = ws.receive_json(timeout=30.0)
                    
                    if message["type"] == "user_joined":
                        print(f"{message['username']} joined")
                    elif message["type"] == "message":
                        print(f"{message['user']}: {message['text']}")
                    elif message["type"] == "user_left":
                        print(f"{message['username']} left")
                        
                except TimeoutError:
                    # Keep connection alive with manual ping if needed
                    pong = ws.ping()
                    pong.wait(timeout=5.0)
                except Exception as e:
                    print(f"Error: {e}")
                    break
            
            # Leave gracefully
            ws.send_json({"type": "leave"})
            ws.close(1000, "Goodbye")

if __name__ == "__main__":
    main()
```

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

### Integrating AsyncWebSocketClient into a Service SDK

Source: https://github.com/frankie567/httpx-ws/blob/main/docs/usage/class_based_client.md

This example shows how to use `AsyncWebSocketClient` within a service SDK, particularly for dependency injection. The `MyServiceSDK` class accepts an `AsyncWebSocketClient` in its constructor and uses it to manage WebSocket connections.

```python
import httpx
from httpx_ws import AsyncWebSocketClient

# Your service SDK
class MyServiceSDK:
    def __init__(self, websocket_client: AsyncWebSocketClient) -> None:
        self._websocket_client = websocket_client

    async def hello(self) -> str:
        async with self._websocket_client.connect("ws://localhost:8000/ws") as ws:
            await ws.send_text("Hello!")
            response = await ws.receive_text()
            return response

# Usage example
async def main() -> None:
    async with httpx.AsyncClient() as client:
        ws_client = AsyncWebSocketClient(client)

        service = MyServiceSDK(ws_client)
        response = await service.hello()
        print(f"Received: {response}")
```

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

### Manual Ping with Sync Session

Source: https://github.com/frankie567/httpx-ws/blob/main/_autodocs/api-reference/ping-manager.md

Demonstrates sending a ping and waiting for a pong response using a synchronous session. Includes examples with and without a custom payload, and with a timeout for the pong response.

```python
from httpx_ws import connect_ws

with connect_ws("ws://localhost/ws") as ws:
    # Send a ping and wait for pong
    pong_event = ws.ping()
    pong_event.wait()  # Blocks until pong received
    print("Pong received")
    
    # Ping with custom payload
    custom_id = b"my_ping_123"
    pong_event = ws.ping(payload=custom_id)
    if pong_event.wait(timeout=5.0):
        print("Pong received within 5 seconds")
    else:
        print("Pong timeout")
```

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

### JSON Transmission Examples

Source: https://github.com/frankie567/httpx-ws/blob/main/_autodocs/types.md

Demonstrates how to send and receive JSON data using different frame types. The default mode for sending and receiving is 'text'.

```python
# Send JSON as text (default)
ws.send_json({"key": "value"})
```

```python
# Send JSON as binary
ws.send_json({"key": "value"}, mode="binary")
```

```python
# Receive JSON from text frames
data = ws.receive_json(mode="text")
```

```python
# Receive JSON from binary frames
data = ws.receive_json(mode="binary")
```

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

### Handling WebSocket Disconnect Exception

Source: https://github.com/frankie567/httpx-ws/blob/main/_autodocs/errors.md

This example demonstrates how to catch and handle the WebSocketDisconnect exception. It shows how to check the close code and reason to determine the cause of the disconnection.

```python
from httpx_ws import aconnect_ws, WebSocketDisconnect

async def chat():
    try:
        async with aconnect_ws("ws://localhost/chat") as ws:
            while True:
                message = await ws.receive_text()
                print(f"Server: {message}")
    except WebSocketDisconnect as e:
        if e.code == 1000:
            print("Server closed normally")
        elif e.code == 1001:
            print("Server is shutting down")
        else:
            print(f"Disconnected: code={e.code}, reason={e.reason}")
```

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

### Sync WebSocket Text Exchange

Source: https://github.com/frankie567/httpx-ws/blob/main/_autodocs/README.md

Use `connect_ws` for synchronous text-based communication with a WebSocket server. This example sends a text message and receives a text response.

```python
from httpx_ws import connect_ws

# Simple text exchange
with connect_ws("ws://localhost:8000/ws") as ws:
    ws.send_text("Hello, Server!")
    response = ws.receive_text()
    print(f"Server said: {response}")
```

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

### Async Chat Application Example

Source: https://github.com/frankie567/httpx-ws/blob/main/_autodocs/api-reference/async-websocket-client.md

This snippet shows a full asynchronous chat client. It connects to a WebSocket server, sends join and message events, receives and processes incoming messages, and handles disconnections and graceful leave operations. Requires an active WebSocket server at ws://localhost:8000/chat.

```python
import asyncio
import httpx
from httpx_ws import AsyncWebSocketClient, WebSocketDisconnect

async def chat_main():
    async with httpx.AsyncClient() as client:
        ws_client = AsyncWebSocketClient(client)
        async with ws_client.connect("ws://localhost:8000/chat") as ws:
            # Join the chat
            await ws.send_json({
                "type": "join",
                "username": "alice"
            })
            
            # Handle incoming messages
            try:
                while True:
                    message = await ws.receive_json(timeout=30.0)
                    
                    if message["type"] == "user_joined":
                        print(f"{message['username']} joined")
                    elif message["type"] == "message":
                        print(f"{message['user']}: {message['text']}")
                    elif message["type"] == "user_left":
                        print(f"{message['username']} left")
                        
            except TimeoutError:
                print("No messages in 30 seconds")
            except WebSocketDisconnect as e:
                print(f"Disconnected: {e.reason}")
            finally:
                # Leave gracefully
                try:
                    await ws.send_json({"type": "leave"})
                    await ws.close(1000, "Goodbye")
                except Exception:
                    pass

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

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

### Create Ping and Get Event

Source: https://github.com/frankie567/httpx-ws/blob/main/_autodocs/api-reference/ping-manager.md

Creates a new ping, returning its ID and a threading.Event to signal when the corresponding pong is received. Use this to initiate a ping-pong exchange.

```python
ping_id, pong_event = ping_manager.create()
# Send ping with this ID
# Later, when pong arrives:
pong_event.wait()  # Blocks until pong received
```

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

### Testing Binary WebSocket Protocol

Source: https://github.com/frankie567/httpx-ws/blob/main/_autodocs/api-reference/asgi-websocket-transport.md

Tests sending and receiving binary data over a WebSocket connection. The example app echoes received bytes, and the test verifies this behavior.

```python
import httpx
from httpx_ws import aconnect_ws
from httpx_ws.transport import ASGIWebSocketTransport

async def binary_app(scope, receive, send):
    assert scope["type"] == "websocket"
    await send({"type": "websocket.accept"})
    
    message = await receive()
    if message["type"] == "websocket.receive":
        data = message.get("bytes", b"")
        await send({
            "type": "websocket.send",
            "bytes": b"Received: " + data
        })

async def test():
    async with ASGIWebSocketTransport(binary_app) as transport:
        async with httpx.AsyncClient(transport=transport) as client:
            async with aconnect_ws("ws://localhost/", client=client) as ws:
                await ws.send_bytes(b"\x00\x01\x02")
                response = await ws.receive_bytes()
                assert response == b"Received: \x00\x01\x02"

import asyncio
asyncio.run(test())
```

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

### AsyncWebSocketSession Context Manager Usage

Source: https://github.com/frankie567/httpx-ws/blob/main/_autodocs/api-reference/async-websocket-session.md

Use AsyncWebSocketSession as an asynchronous context manager with 'async with' for proper session setup and teardown. It handles stream creation, background tasks, and automatic connection closure.

```python
async with AsyncWebSocketSession(stream) as ws:
    await ws.send_text("Hello")
    message = await ws.receive_text()
    # Connection automatically closed on exit
```

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

### Testing ASGI WebSocket Endpoints

Source: https://github.com/frankie567/httpx-ws/blob/main/_autodocs/MODULE_OVERVIEW.md

Use ASGIWebSocketTransport to test ASGI WebSocket endpoints without a server. This example shows how to set up the transport and an httpx client to interact with a simulated WebSocket connection.

```python
async with ASGIWebSocketTransport(app) as transport:
    async with httpx.AsyncClient(transport=transport) as client:
        async with aconnect_ws("ws://test/", client=client) as ws:
            # Test code here
```

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

### Manual Ping with Async Session

Source: https://github.com/frankie567/httpx-ws/blob/main/_autodocs/api-reference/ping-manager.md

Shows how to send a ping and await a pong response in an asynchronous context. Covers basic ping, ping with a custom payload, and handling timeouts using anyio.

```python
from httpx_ws import aconnect_ws

async def ping_example():
    async with aconnect_ws("ws://localhost/ws") as ws:
        # Send a ping and wait for pong
        pong_event = await ws.ping()
        await pong_event.wait()
        print("Pong received")
        
        # Ping with custom payload
        custom_id = b"my_ping_456"
        pong_event = await ws.ping(payload=custom_id)
        try:
            async with anyio.fail_after(5.0):
                await pong_event.wait()
            print("Pong received within 5 seconds")
        except TimeoutError:
            print("Pong timeout")
```

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

### Connect to WebSocket (Sync)

Source: https://github.com/frankie567/httpx-ws/blob/main/docs/usage/quickstart.md

Use `connect_ws` for synchronous WebSocket connections. Requires importing `connect_ws` from `httpx_ws`.

```python
from httpx_ws import connect_ws

with connect_ws("http://localhost:8000/ws") as ws:
    message = ws.receive_text()
    print(message)
    ws.send_text("Hello!")
```

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

### ASGIWebSocketTransport Context Manager

Source: https://github.com/frankie567/httpx-ws/blob/main/_autodocs/api-reference/asgi-websocket-transport.md

Provides asynchronous context management for the ASGIWebSocketTransport, handling setup and teardown. It is commonly used with httpx.AsyncClient for WebSocket testing.

```APIDOC
## ASGIWebSocketTransport Context Manager

### Description
Provides asynchronous context management for the ASGIWebSocketTransport, handling setup and teardown. It is commonly used with httpx.AsyncClient for WebSocket testing.

### Methods
- `__aenter__`
- `__aexit__`

### Returns
The transport instance.

### Example
```python
async with ASGIWebSocketTransport(app) as transport:
    async with httpx.AsyncClient(transport=transport) as client:
        async with aconnect_ws("ws://localhost/ws", client=client) as ws:
            await ws.send_text("Hello")
            response = await ws.receive_text()
```
```

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

### Receive Raw WebSocket Event with Timeout

Source: https://github.com/frankie567/httpx-ws/blob/main/_autodocs/api-reference/async-websocket-session.md

Use `ws.receive()` to get any WebSocket event, with an optional timeout. Handles `TimeoutError`, `WebSocketDisconnect`, and `WebSocketNetworkError`.

```python
async with ws:
    try:
        event = await ws.receive(timeout=5.0)
        if isinstance(event, wsproto.events.TextMessage):
            print(f"Text: {event.data}")
    except TimeoutError:
        print("No message within 5 seconds")
    except WebSocketDisconnect:
        print("Connection closed")
```

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

### Testing ASGI App with HTTPX-WS

Source: https://github.com/frankie567/httpx-ws/blob/main/docs/usage/asgi.md

Demonstrates how to test an ASGI application's HTTP and WebSocket endpoints using httpx-ws. Requires passing the httpx client instance to aconnect_ws.

```python
import httpx
from httpx_ws import aconnect_ws
from httpx_ws.transport import ASGIWebSocketTransport

async with httpx.AsyncClient(transport=ASGIWebSocketTransport(app)) as client:
    http_response = await client.get("http://server/http")
    assert http_response.status_code == 200

    async with aconnect_ws("http://server/ws", client) as ws:
        message = await ws.receive_text()
        assert message == "Hello World!"
```

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

### HTTPX-WS Exception Hierarchy

Source: https://github.com/frankie567/httpx-ws/blob/main/_autodocs/INDEX.md

Illustrates the exception hierarchy for HTTPX-WS, starting from a base exception and including specific error types like WebSocketUpgradeError and WebSocketDisconnect.

```python
Exception Hierarchy:
└── HTTPXWSException (base)
    ├── WebSocketUpgradeError(response: httpx.Response)
    ├── WebSocketDisconnect(code: int, reason: str | None)
    ├── WebSocketInvalidTypeReceived(event: wsproto.events.Event)
    └── WebSocketNetworkError()
```

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

### Synchronous WebSocket Connection Establishment

Source: https://github.com/frankie567/httpx-ws/blob/main/_autodocs/MODULE_OVERVIEW.md

Illustrates the sequence of operations for establishing a synchronous WebSocket connection. This involves using `httpx.Client.stream()` to initiate the connection and `WebSocketSession` to manage the session.

```text
connect_ws()
  ↓
httpx.Client.stream() → GET /ws with upgrade headers
  ↓
HTTP 101 Switching Protocols response
  ↓
WebSocketSession(stream)
  ↓
__enter__()
  ├─ Start _background_receive thread
  └─ Start _background_keepalive_ping thread (optional)
  ↓
Ready for send_*() and receive_*() calls
```

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

### Type Hints for Async WebSocket Clients

Source: https://github.com/frankie567/httpx-ws/blob/main/_autodocs/README.md

Demonstrates the type hints available for asynchronous WebSocket clients and sessions, including the main client, session, and generator types.

```python
from httpx_ws import AsyncWebSocketClient, AsyncWebSocketSession
from typing import AsyncGenerator
import httpx

client: httpx.AsyncClient
ws_client: AsyncWebSocketClient
session: AsyncWebSocketSession
```

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

### Asynchronous WebSocket Connection Establishment

Source: https://github.com/frankie567/httpx-ws/blob/main/_autodocs/MODULE_OVERVIEW.md

Details the process for establishing an asynchronous WebSocket connection using `httpx.AsyncClient.stream()`. It highlights the use of `AsyncWebSocketSession` and the creation of background tasks.

```text
aconnect_ws()
  ↓
httpx.AsyncClient.stream() → GET /ws with upgrade headers
  ↓
HTTP 101 Switching Protocols response
  ↓
AsyncWebSocketSession(stream)
  ↓
__aenter__()
  ├─ Create anyio memory object streams
  ├─ Create anyio task group
  ├─ Start _background_receive task
  └─ Start _background_keepalive_ping task (optional)
  ↓
Ready for async send_*() and receive_*() calls
```

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

### WebSocket JSON Data Exchange

Source: https://github.com/frankie567/httpx-ws/blob/main/_autodocs/README.md

Send and receive JSON data over WebSockets using `send_json` and `receive_json`. This example subscribes to a channel and processes incoming price data.

```python
from httpx_ws import connect_ws

with connect_ws("ws://localhost:8000/api") as ws:
    # Send JSON
    ws.send_json({"action": "subscribe", "channel": "prices"})
    
    # Receive JSON
    while True:
        data = ws.receive_json(timeout=30.0)
        print(f"Price: {data['price']}")
```

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

### Async WebSocket Text Exchange

Source: https://github.com/frankie567/httpx-ws/blob/main/_autodocs/README.md

Utilize `aconnect_ws` for asynchronous text-based WebSocket communication. This example demonstrates sending and receiving text messages within an async function.

```python
from httpx_ws import aconnect_ws

async def main():
    async with aconnect_ws("ws://localhost:8000/ws") as ws:
        await ws.send_text("Hello, Server!")
        response = await ws.receive_text()
        print(f"Server said: {response}")

import asyncio
asyncio.run(main())
```

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

### Connect with Subprotocols (Async)

Source: https://github.com/frankie567/httpx-ws/blob/main/docs/usage/subprotocols.md

This snippet demonstrates establishing an asynchronous WebSocket connection, specifying supported subprotocols. The server's chosen subprotocol can be accessed through `ws.subprotocol`.

```python
from httpx_ws import aconnect_ws

async with aconnect_ws(
    "http://localhost:8000/ws",
    subprotocols=["my_protocol_1", "my_protocol_2"],
) as ws:
    print(f"Accepted subprotocol {ws.subprotocol}")
```

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

### AsyncWebSocketClient Constructor

Source: https://github.com/frankie567/httpx-ws/blob/main/_autodocs/api-reference/async-websocket-client.md

Initializes an AsyncWebSocketClient instance. It requires an httpx.AsyncClient and allows configuration of message size, queue size, keep-alive intervals, and the session class.

```python
def __init__(
    self,
    client: httpx.AsyncClient,
    *,
    max_message_size_bytes: int = DEFAULT_MAX_MESSAGE_SIZE_BYTES,
    queue_size: int = DEFAULT_QUEUE_SIZE,
    keepalive_ping_interval_seconds: float | None = DEFAULT_KEEPALIVE_PING_INTERVAL_SECONDS,
    keepalive_ping_timeout_seconds: float | None = DEFAULT_KEEPALIVE_PING_TIMEOUT_SECONDS,
    session_class: type[AsyncSession] = AsyncWebSocketSession,
) -> None:
```

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

### Connect to WebSocket with Custom Client (Sync)

Source: https://github.com/frankie567/httpx-ws/blob/main/docs/usage/quickstart.md

Pass an `httpx.Client` instance to `connect_ws` for customized synchronous connections and connection pooling.

```python
import httpx
from httpx_ws import connect_ws

with httpx.Client() as client:
    with connect_ws("http://localhost:8000/ws", client) as ws:
        message = ws.receive_text()
        print(message)
        ws.send_text("Hello!")
```

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

### Exception Hierarchy for HTTPXWS

Source: https://github.com/frankie567/httpx-ws/blob/main/_autodocs/MODULE_OVERVIEW.md

Defines the exception hierarchy for the httpx-ws library, starting from a base exception and including specific errors for WebSocket upgrades, disconnections, invalid data, and network issues.

```python
HTTPXWSException (base)
├── WebSocketUpgradeError
├── WebSocketDisconnect
├── WebSocketInvalidTypeReceived
└── WebSocketNetworkError
```

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

### Connect with Subprotocols (Sync)

Source: https://github.com/frankie567/httpx-ws/blob/main/docs/usage/subprotocols.md

Use this snippet to establish a synchronous WebSocket connection with a list of supported subprotocols. The accepted subprotocol is available via `ws.subprotocol`.

```python
from httpx_ws import connect_ws

with connect_ws(
    "http://localhost:8000/ws",
    subprotocols=["my_protocol_1", "my_protocol_2"],
) as ws:
    print(f"Accepted subprotocol {ws.subprotocol}")
```

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

### Send and Receive Messages Concurrently in WebSocket Client

Source: https://github.com/frankie567/httpx-ws/blob/main/_autodocs/api-reference/async-websocket-client.md

This example shows how to concurrently send and receive messages over a single WebSocket connection. It uses `asyncio.gather` to run sender and receiver tasks simultaneously.

```python
import asyncio
from httpx_ws import aconnect_ws

async def echo_client():
    async with aconnect_ws("ws://localhost:8000/echo") as ws:
        async def send_messages():
            for i in range(10):
                await ws.send_json({"count": i})
                await asyncio.sleep(1.0)
                await ws.send_text("ping")
                await asyncio.sleep(1.0)
        
        async def receive_messages():
            while True:
                message = await ws.receive_json()
                print(f"Received: {message}")
        
        await asyncio.gather(
            send_messages(),
            receive_messages(),
        )

asyncio.run(echo_client())
```

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

### WebSocketClient Constructor

Source: https://github.com/frankie567/httpx-ws/blob/main/_autodocs/api-reference/websocket-client.md

Initializes a new WebSocketClient instance. Configure connection parameters like message size, queue size, and keep-alive settings.

```python
def __init__(
    self,
    client: httpx.Client,
    *,
    max_message_size_bytes: int = DEFAULT_MAX_MESSAGE_SIZE_BYTES,
    queue_size: int = DEFAULT_QUEUE_SIZE,
    keepalive_ping_interval_seconds: float | None = DEFAULT_KEEPALIVE_PING_INTERVAL_SECONDS,
    keepalive_ping_timeout_seconds: float | None = DEFAULT_KEEPALIVE_PING_TIMEOUT_SECONDS,
    session_class: type[SyncSession] = WebSocketSession,
) -> None:
```

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

### Pass HTTPX Integration Parameters

Source: https://github.com/frankie567/httpx-ws/blob/main/_autodocs/configuration.md

Configure the underlying HTTPX connection by passing standard parameters like authentication, custom headers, cookies, timeouts, SSL verification settings, and proxies directly to `connect_ws`.

```python
with connect_ws(
    "ws://localhost/ws",
    # Authentication
    auth=("username", "password"),
    
    # Custom headers
    headers={"User-Agent": "MyApp/1.0", "X-Custom": "value"},
    
    # Cookies
    cookies={"session": "abc123"},
    
    # Timeout
    timeout=30.0,
    
    # SSL/TLS verification
    verify=True,  # or path to CA cert
    
    # Proxy
    proxies="http://proxy.example.com:8080",
) as ws:
    pass
```

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

### WebSocketSession Constructor

Source: https://github.com/frankie567/httpx-ws/blob/main/_autodocs/api-reference/websocket-session.md

Initializes a WebSocketSession with network stream and optional configuration parameters.

```python
def __init__(
    self,
    stream: NetworkStream,
    *,
    max_message_size_bytes: int = DEFAULT_MAX_MESSAGE_SIZE_BYTES,
    queue_size: int = DEFAULT_QUEUE_SIZE,
    keepalive_ping_interval_seconds: float | None = DEFAULT_KEEPALIVE_PING_INTERVAL_SECONDS,
    keepalive_ping_timeout_seconds: float | None = DEFAULT_KEEPALIVE_PING_TIMEOUT_SECONDS,
    response: httpx.Response | None = None,
) -> None:
```

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

### Testing WebSocket Connection Rejection

Source: https://github.com/frankie567/httpx-ws/blob/main/_autodocs/api-reference/asgi-websocket-transport.md

Tests how the transport handles a rejected WebSocket connection. The example app explicitly denies the connection with a 403 status code, and the test verifies that a WebSocketUpgradeError is raised with the correct response.

```python
import httpx
from httpx_ws import aconnect_ws
from httpx_ws.transport import ASGIWebSocketTransport
from httpx_ws import WebSocketUpgradeError

async def rejection_app(scope, receive, send):
    # Deny the connection
    await send({
        "type": "websocket.http.response.start",
        "status": 403,
        "headers": [[b"content-type", b"text/plain"]],
    })
    await send({
        "type": "websocket.http.response.body",
        "body": b"Access Denied",
    })

async def test():
    async with ASGIWebSocketTransport(rejection_app) as transport:
        async with httpx.AsyncClient(transport=transport) as client:
            try:
                async with aconnect_ws("ws://localhost/", client=client) as ws:
                    pass
            except WebSocketUpgradeError as e:
                assert e.response.status_code == 403
                print("✓ Rejection handled correctly")

import asyncio
asyncio.run(test())
```

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

### AsyncWebSocketClient Constructor

Source: https://github.com/frankie567/httpx-ws/blob/main/_autodocs/api-reference/async-websocket-client.md

Initializes an AsyncWebSocketClient. This client wraps an HTTPX async client to manage WebSocket connections, offering configurable options for message size, queue size, and keep-alive settings.

```APIDOC
## AsyncWebSocketClient Constructor

### Description
Initializes an `AsyncWebSocketClient` instance. This client is designed to work with an `httpx.AsyncClient` to establish and manage WebSocket connections.

### Parameters
#### Path Parameters
None

#### Query Parameters
None

#### Request Body
None

### Parameters
- **client** (`httpx.AsyncClient`) - Required - The HTTPX async client instance used for establishing the connection.
- **max_message_size_bytes** (`int`) - Optional - The maximum size in bytes for incoming messages. Defaults to `65536`.
- **queue_size** (`int`) - Optional - The size of the queue for received messages. Defaults to `512`.
- **keepalive_ping_interval_seconds** (`float | None`) - Optional - The interval in seconds for sending keep-alive pings. Set to `None` to disable. Defaults to `20.0`.
- **keepalive_ping_timeout_seconds** (`float | None`) - Optional - The timeout in seconds for receiving a pong response after sending a ping. Defaults to `20.0`.
- **session_class** (`type[AsyncSession]`) - Optional - The custom session class to use for WebSocket sessions. Defaults to `AsyncWebSocketSession`.

### Attributes
- **client** (`httpx.AsyncClient`): The underlying HTTPX async client.
- **max_message_size_bytes** (`int`): Configuration for the maximum message size.
- **queue_size** (`int`): Configuration for the received message queue size.
- **keepalive_ping_interval_seconds** (`float | None`): Configuration for the keep-alive ping interval.
- **keepalive_ping_timeout_seconds** (`float | None`): Configuration for the keep-alive pong timeout.
- **session_class** (`type[AsyncSession]`): The async session class used for connections.
```

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

### Import Locations - Exception Types

Source: https://github.com/frankie567/httpx-ws/blob/main/_autodocs/types.md

Demonstrates importing various exception types from the httpx_ws library.

```python
# Exception types
from httpx_ws import (
    HTTPXWSException,
    WebSocketUpgradeError,
    WebSocketDisconnect,
    WebSocketInvalidTypeReceived,
    WebSocketNetworkError,
)
```

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

### WebSocketSession as Context Manager

Source: https://github.com/frankie567/httpx-ws/blob/main/_autodocs/api-reference/websocket-session.md

Utilize `WebSocketSession` as a context manager for automatic connection management. It handles starting background threads for receiving messages and keepalive pings, and ensures the connection is closed upon exiting the `with` block.

```python
with WebSocketSession(stream) as ws:
    ws.send_text("Hello")
    message = ws.receive_text()
    # Connection automatically closed on exit
```

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

### Connect to WebSocket (Async)

Source: https://github.com/frankie567/httpx-ws/blob/main/docs/usage/quickstart.md

Use `aconnect_ws` for asynchronous WebSocket connections. Requires importing `aconnect_ws` from `httpx_ws`.

```python
from httpx_ws import aconnect_ws

async with aconnect_ws("http://localhost:8000/ws") as ws:
    message = await ws.receive_text()
    print(message)
    await ws.send_text("Hello!")
```

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

### Establish Async WebSocket Connection

Source: https://github.com/frankie567/httpx-ws/blob/main/_autodocs/api-reference/async-websocket-client.md

Use this method within an `httpx.AsyncClient` context to establish a WebSocket connection. It yields an `AsyncSession` object for sending and receiving messages. Ensure the server responds with status 101 for a successful upgrade.

```python
async with httpx.AsyncClient() as client:
    ws_client = AsyncWebSocketClient(client)
    async with ws_client.connect("ws://localhost:8000/ws") as ws:
        await ws.send_text("Hello!")
        message = await ws.receive_text()
        print(message)
```

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

### Run Linting

Source: https://github.com/frankie567/httpx-ws/blob/main/AGENTS.md

Execute the linting and type-checking process for the project. Ensure code quality and consistency.

```bash
just lint
```

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

### connect

Source: https://github.com/frankie567/httpx-ws/blob/main/_autodocs/api-reference/async-websocket-client.md

Establishes an asynchronous WebSocket connection to a given URL. It acts as an async context manager, yielding an `AsyncSession` upon successful connection.

```APIDOC
## connect

### Description
Establish an async WebSocket connection to the given URL.

### Method
`async def connect(self, url: str, subprotocols: list[str] | None = None, **kwargs: typing.Any)`

### Parameters
#### Path Parameters
None

#### Query Parameters
None

#### Request Body
None

### Parameters
- `url` (`str`): WebSocket URL (e.g., `"ws://localhost:8000/chat"`).
- `subprotocols` (`list[str] | None`): Optional list of subprotocols to negotiate. Defaults to `None`.
- `**kwargs`: Additional arguments passed to HTTPX's `stream()` method (headers, auth, timeout, etc.).

### Returns
Async context manager yielding an `AsyncWebSocketSession`.

### Raises
- `WebSocketUpgradeError` — If the server doesn't respond with status 101.
- `httpx.RequestError` — If the HTTP request fails.

### Example
```python
async with httpx.AsyncClient() as client:
    ws_client = AsyncWebSocketClient(client)
    async with ws_client.connect("ws://localhost:8000/ws") as ws:
        await ws.send_text("Hello!")
        message = await ws.receive_text()
        print(message)
```
```

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

### Instantiating WebSocketClient with Custom Session

Source: https://github.com/frankie567/httpx-ws/blob/main/_autodocs/MODULE_OVERVIEW.md

Shows how to instantiate the WebSocketClient with a custom session class, leveraging the generic type variable.

```python
client = WebSocketClient(httpx.Client(), session_class=CustomSession)
```

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

### Connection Parameters

Source: https://github.com/frankie567/httpx-ws/blob/main/_autodocs/INDEX.md

Parameters used for establishing WebSocket connections.

```APIDOC
## Connection Parameters

### Description
Parameters used for establishing WebSocket connections.

### Parameters
#### Connection Parameters
- **url** (str) - Required - The WebSocket URL to connect to.
- **subprotocols** (list[str] | None) - Optional - A list of subprotocols to negotiate.
- **max_message_size_bytes** (int) - Optional - The maximum size of messages in bytes. Defaults to 65536.
- **queue_size** (int) - Optional - The size of the internal queue. Defaults to 512.
- **keepalive_ping_interval_seconds** (float | None) - Optional - The interval in seconds for sending keep-alive pings. Defaults to 20.0.
- **keepalive_ping_timeout_seconds** (float | None) - Optional - The timeout in seconds for receiving keep-alive pings. Defaults to 20.0.
- **session_class** (type) - Optional - The session class to use (e.g., WebSocketSession or AsyncWebSocketSession).
```

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

### WebSocketClient Constructor

Source: https://github.com/frankie567/httpx-ws/blob/main/_autodocs/api-reference/websocket-client.md

Initializes a new WebSocketClient instance. This client wraps an HTTPX client to manage WebSocket connections.

```APIDOC
## WebSocketClient Constructor

### Description
Initializes a new WebSocketClient instance. This client wraps an HTTPX client to manage WebSocket connections.

### Parameters
#### Path Parameters
None

#### Query Parameters
None

#### Request Body
None

### Parameters
- **client** (`httpx.Client`) - Required - HTTPX client for establishing connections
- **max_message_size_bytes** (`int`) - Optional - Maximum message size to receive (default: `65536`)
- **queue_size** (`int`) - Optional - Size of received message queue (default: `512`)
- **keepalive_ping_interval_seconds** (`float | None`) - Optional - Automatic ping interval (seconds). `None` disables. (default: `20.0`)
- **keepalive_ping_timeout_seconds** (`float | None`) - Optional - Timeout for pong response (default: `20.0`)
- **session_class** (`type[SyncSession]`) - Optional - Custom session class to use (default: `WebSocketSession`)

### Attributes
- **client** (`httpx.Client`) - The HTTPX client instance
- **max_message_size_bytes** (`int`) - Maximum message size configuration
- **queue_size** (`int`) - Queue size configuration
- **keepalive_ping_interval_seconds** (`float | None`) - Ping interval configuration
- **keepalive_ping_timeout_seconds** (`float | None`) - Pong timeout configuration
- **session_class** (`type[SyncSession]`) - Session class to instantiate on connect
```

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

### Main API Imports

Source: https://github.com/frankie567/httpx-ws/blob/main/_autodocs/INDEX.md

Import core components for WebSocket communication, including session and connection functions.

```python
# Main API
from httpx_ws import (
    WebSocketSession,
    AsyncWebSocketSession,
    WebSocketClient,
    AsyncWebSocketClient,
    connect_ws,
    aconnect_ws,
    JSONMode,
)
```

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

### Asynchronous WebSocket Client Connection

Source: https://github.com/frankie567/httpx-ws/blob/main/docs/usage/class_based_client.md

This snippet demonstrates establishing an asynchronous WebSocket connection. It requires an `httpx.AsyncClient` and `AsyncWebSocketClient`. The connection is managed using `async with`.

```python
import httpx
from httpx_ws import AsyncWebSocketClient

async with httpx.AsyncClient() as client:
    ws_client = AsyncWebSocketClient(client)
    async with ws_client.connect("http://localhost:8000/ws") as ws:
        message = await ws.receive_text()
        print(message)
        await ws.send_text("Hello!")
```

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

### connect_ws

Source: https://github.com/frankie567/httpx-ws/blob/main/_autodocs/INDEX.md

Factory function to establish a synchronous WebSocket connection.

```APIDOC
## connect_ws

### Description
Factory function to establish a synchronous WebSocket connection.

### Signature
`connect_ws(url, client, ...) -> ContextManager[WebSocketSession]`
```

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

### Type Hierarchy

Source: https://github.com/frankie567/httpx-ws/blob/main/_autodocs/types.md

Illustrates the inheritance hierarchy for events from wsproto and httpxws exceptions.

```text
Event (from wsproto)
├── TextMessage
├── BytesMessage
├── Ping
├── Pong
├── CloseConnection
└── ...

HTTPXWSException
├── WebSocketUpgradeError
├── WebSocketDisconnect
├── WebSocketInvalidTypeReceived
└── WebSocketNetworkError
```

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

### WebSocketSession Constructor

Source: https://github.com/frankie567/httpx-ws/blob/main/_autodocs/api-reference/websocket-session.md

Initializes a new WebSocketSession instance. This class acts as a synchronous context manager for an opened WebSocket connection.

```APIDOC
## WebSocketSession Constructor

### Description
Initializes a new WebSocketSession instance. This class acts as a synchronous context manager for an opened WebSocket connection.

### Parameters
#### Parameters
- **stream** (`httpcore.NetworkStream`) - Required - The underlying network stream for WebSocket communication
- **max_message_size_bytes** (`int`) - Optional - Maximum size of messages to receive from server (Default: `65536`)
- **queue_size** (`int`) - Optional - Size of the queue for received messages before consumer consumes them (Default: `512`)
- **keepalive_ping_interval_seconds** (`float | None`) - Optional - Interval for automatic ping events. Set to `None` to disable. (Default: `20.0`)
- **keepalive_ping_timeout_seconds** (`float | None`) - Optional - Maximum wait time for pong response. Set to `None` to disable timeout. (Default: `20.0`)
- **response** (`httpx.Response | None`) - Optional - The WebSocket upgrade response from the server (Default: `None`)
```

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

### Import Locations - Main Types

Source: https://github.com/frankie567/httpx-ws/blob/main/_autodocs/types.md

Shows how to import the main JSONMode type from the httpx_ws library.

```python
# Main types
from httpx_ws import JSONMode
```

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

### Test ASGI App with WebSocket Transport

Source: https://github.com/frankie567/httpx-ws/blob/main/_autodocs/INDEX.md

Demonstrates testing an ASGI application by using `ASGIWebSocketTransport` and `httpx.AsyncClient` to simulate WebSocket interactions.

```python
from httpx_ws import aconnect_ws
from httpx_ws.transport import ASGIWebSocketTransport
import httpx

async def test():
    async with ASGIWebSocketTransport(app) as transport:
        async with httpx.AsyncClient(transport=transport) as client:
            async with aconnect_ws("ws://localhost/ws", client=client) as ws:
                await ws.send_text("test")
                assert await ws.receive_text() == "test"
```

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

### Handle Message Receive Timeouts

Source: https://github.com/frankie567/httpx-ws/blob/main/_autodocs/errors.md

Demonstrates how to set a timeout for receiving messages and handle a TimeoutError if no message arrives within the specified duration. Includes sending a ping to keep the connection alive.

```python
from httpx_ws import connect_ws

with connect_ws("ws://localhost/ws") as ws:
    try:
        message = ws.receive_text(timeout=5.0)
    except TimeoutError:
        print("No message received in 5 seconds")
        # Send a ping to keep connection alive
        pong = ws.ping()
        if pong.wait(timeout=5.0):
            print("Pong received")
```

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

### AsyncWebSocketSession Constructor

Source: https://github.com/frankie567/httpx-ws/blob/main/_autodocs/api-reference/async-websocket-session.md

Initializes an AsyncWebSocketSession with the underlying network stream and various configuration options for message handling and keep-alive pings.

```python
def __init__(
    self,
    stream: AsyncNetworkStream,
    *,
    max_message_size_bytes: int = DEFAULT_MAX_MESSAGE_SIZE_BYTES,
    queue_size: int = DEFAULT_QUEUE_SIZE,
    keepalive_ping_interval_seconds: float | None = DEFAULT_KEEPALIVE_PING_INTERVAL_SECONDS,
    keepalive_ping_timeout_seconds: float | None = DEFAULT_KEEPALIVE_PING_TIMEOUT_SECONDS,
    response: httpx.Response | None = None,
) -> None:
```

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

### AsyncPingManager.create

Source: https://github.com/frankie567/httpx-ws/blob/main/_autodocs/api-reference/ping-manager.md

Creates a new ping and returns its ID and an anyio event to wait on. This is useful for initiating a ping-pong sequence.

```APIDOC
## create

### Description
Create a new ping and return its ID and an anyio event to wait on.

### Method Signature
`create(self, ping_id: bytes | None = None) -> tuple[bytes, anyio.Event]`

### Parameters
#### `ping_id`
- **Type**: `bytes | None`
- **Optional**: Yes
- **Description**: Optional custom ping ID. If `None`, a random ID is generated.

### Returns
- **Type**: `tuple[bytes, anyio.Event]`
- **Description**: A tuple containing the ping ID (bytes) and an anyio event that is set when the corresponding pong is received.

### Example
```python
ping_id, pong_event = ping_manager.create()
# Send ping with this ID
# Later, when pong arrives:
await pong_event.wait()  # Awaits until pong received
```
```

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

### Balanced Configuration Profile

Source: https://github.com/frankie567/httpx-ws/blob/main/_autodocs/configuration.md

This is the default configuration, balancing message size, queue size, and keepalive intervals for general use.

```python
with connect_ws(
    url,
    max_message_size_bytes=65 * 1024,   # 65 KiB
    queue_size=512,
    keepalive_ping_interval_seconds=20.0,
    keepalive_ping_timeout_seconds=20.0,
) as ws:
    pass
```

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

### Synchronous WebSocket Client Connection

Source: https://github.com/frankie567/httpx-ws/blob/main/docs/usage/class_based_client.md

Use this snippet to establish a synchronous WebSocket connection. It requires an existing `httpx.Client` instance and the `WebSocketClient` from `httpx_ws`. The connection is managed using a context manager.

```python
import httpx
from httpx_ws import WebSocketClient

with httpx.Client() as client:
    ws_client = WebSocketClient(client)
    with ws_client.connect("http://localhost:8000/ws") as ws:
        message = ws.receive_text()
        print(message)
        ws.send_text("Hello!")
```

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

### Wait for Pong with Timeout (Async)

Source: https://github.com/frankie567/httpx-ws/blob/main/_autodocs/api-reference/ping-manager.md

Illustrates how to handle pong timeouts in the asynchronous version using `anyio.fail_after`. If a pong is not received within the specified duration, a `TimeoutError` is caught.

```python
pong_event = await ws.ping()
try:
    async with anyio.fail_after(5.0):
        await pong_event.wait()
except TimeoutError:
    print("Pong timeout")
```

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

### Connect to WebSocket with Custom Client (Async)

Source: https://github.com/frankie567/httpx-ws/blob/main/docs/usage/quickstart.md

Pass an `httpx.AsyncClient` instance to `aconnect_ws` for customized asynchronous connections and connection pooling.

```python
import httpx
from httpx_ws import aconnect_ws

async with httpx.AsyncClient() as client:
    async with aconnect_ws("http://localhost:8000/ws", client) as ws:
        message = await ws.receive_text()
        print(message)
        await ws.send_text("Hello!")
```

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

### WebSocketClient.connect

Source: https://github.com/frankie567/httpx-ws/blob/main/_autodocs/api-reference/websocket-client.md

Establishes a WebSocket connection using a context manager. It allows for custom subprotocols and passes additional arguments to the underlying HTTPX stream method.

```APIDOC
## WebSocketClient.connect

### Description
Establishes a WebSocket connection to the given URL.

### Method Signature
```python
@contextlib.contextmanager
def connect(
    self,
    url: str,
    subprotocols: list[str] | None = None,
    **kwargs: typing.Any,
) -> typing.Generator[SyncSession, None, None]:
```

### Parameters
#### Path Parameters
- `url` (str) - Required - WebSocket URL (e.g., "ws://localhost:8000/chat").
- `subprotocols` (list[str] | None) - Optional - Optional list of subprotocols to negotiate. Defaults to `None`.
- `**kwargs` (typing.Any) - Additional arguments passed to HTTPX's `stream()` method (headers, auth, timeout, etc.).

### Returns
Context manager yielding a `WebSocketSession`.

### Raises
- `WebSocketUpgradeError` — If the server doesn't respond with status 101.
- `httpx.RequestError` — If the HTTP request fails.

### Example
```python
with httpx.Client() as client:
    ws_client = WebSocketClient(client)
    with ws_client.connect("ws://localhost:8000/ws") as ws:
        ws.send_text("Hello!")
        message = ws.receive_text()
        print(message)
```
```

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

### Testing WebSocket Subprotocol Negotiation

Source: https://github.com/frankie567/httpx-ws/blob/main/_autodocs/api-reference/asgi-websocket-transport.md

Demonstrates testing subprotocol negotiation during WebSocket connection establishment. The test specifies desired subprotocols and asserts that the chosen subprotocol matches expectations.

```python
import httpx
from httpx_ws import aconnect_ws
from httpx_ws.transport import ASGIWebSocketTransport

async def subprotocol_app(scope, receive, send):
    subprotocols = scope.get("subprotocols", [])
    chosen = "chat" if "chat" in subprotocols else None
    
    await send({
        "type": "websocket.accept",
        "subprotocol": chosen,
    })
    
    # ... handle messages ...

async def test():
    async with ASGIWebSocketTransport(subprotocol_app) as transport:
        async with httpx.AsyncClient(transport=transport) as client:
            async with aconnect_ws(
                "ws://localhost/",
                client=client,
                subprotocols=["chat", "superchat"],
            ) as ws:
                assert ws.subprotocol == "chat"

import asyncio
asyncio.run(test())
```

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

### WebSocket Connection Factory Functions

Source: https://github.com/frankie567/httpx-ws/blob/main/_autodocs/INDEX.md

Presents factory functions for establishing WebSocket connections, one for sync and one for async contexts.

```python
Factory Functions:
├── connect_ws(url, client, ...) → ContextManager[WebSocketSession]
└── aconnect_ws(url, client, ...) → AsyncContextManager[AsyncWebSocketSession]
```

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

### Convenient WebSocket Connection Factory

Source: https://github.com/frankie567/httpx-ws/blob/main/_autodocs/api-reference/websocket-client.md

Use the `connect_ws` factory function for a simplified way to create a client and connect to a WebSocket URL in one step. It accepts various configuration options for message size, keepalive, and subprotocols.

```python
from httpx_ws import connect_ws

with connect_ws("ws://localhost:8000/ws") as ws:
    ws.send_text("Hello!")
    message = ws.receive_text()
    print(message)
```

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

### Aggressive Keepalive Configuration for Production

Source: https://github.com/frankie567/httpx-ws/blob/main/_autodocs/INDEX.md

Configure aggressive keepalive detection for production environments to ensure timely detection of connection issues.

```python
connect_ws(
    url,
    keepalive_ping_interval_seconds=10.0,
    keepalive_ping_timeout_seconds=5.0,
)
```

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

### Import Locations - ASGI Types

Source: https://github.com/frankie567/httpx-ws/blob/main/_autodocs/types.md

Illustrates importing ASGI-related types such as Scope, Message, Receive, Send, and ASGIApp from the httpx_ws.transport module.

```python
# ASGI types (transport module)
from httpx_ws.transport import (
    Scope,
    Message,
    Receive,
    Send,
    ASGIApp,
)
```

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

### AsyncWebSocketClient

Source: https://github.com/frankie567/httpx-ws/blob/main/_autodocs/INDEX.md

Async client factory for establishing asynchronous WebSocket connections. Initializes with an httpx.AsyncClient and offers an async connect method for creating an AsyncWebSocketSession.

```APIDOC
## AsyncWebSocketClient

### Description
Async client factory for establishing asynchronous WebSocket connections. Initializes with an httpx.AsyncClient and offers an async connect method for creating an AsyncWebSocketSession.

### Methods
- `__init__(httpx.AsyncClient, ...) -> None`
- `async connect(url, subprotocols, **kwargs) -> AsyncContextManager[AsyncWebSocketSession]`
```

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

### aconnect_ws()

Source: https://github.com/frankie567/httpx-ws/blob/main/_autodocs/README.md

Asynchronous factory function to establish a WebSocket connection and return an async session context manager.

```APIDOC
## aconnect_ws()

### Description
Asynchronous factory function to establish a WebSocket connection.

### Parameters
- `url` (str): The WebSocket URL to connect to.
- `client` (AsyncWebSocketClient): The async client instance to use.
- `config` (Any): Configuration options.

### Returns
An async context manager yielding an `AsyncWebSocketSession` object.
```

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

### Establish Async WebSocket Connection

Source: https://github.com/frankie567/httpx-ws/blob/main/_autodocs/INDEX.md

Use `aconnect_ws` for asynchronous WebSocket connections. This requires an `async` context.

```python
from httpx_ws import aconnect_ws

# Simplest usage
async with aconnect_ws("ws://localhost:8000/ws") as ws:
    await ws.send_text("Hello")
    response = await ws.receive_text()
    print(response)
```

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

### Async Manual Ping/Pong

Source: https://github.com/frankie567/httpx-ws/blob/main/_autodocs/INDEX.md

Send a ping frame asynchronously and wait for the pong response using `await ws.ping()` and `anyio.fail_after` for timeout.

```python
from httpx_ws import aconnect_ws

async with aconnect_ws("ws://localhost:8000/ws") as ws:
    pong_event = await ws.ping()
    try:
        async with anyio.fail_after(5.0):
            await pong_event.wait()
        print("Pong received")
    except TimeoutError:
        print("Pong timeout")
```