aiohttp_ws_client

What the client does

Quick start

Install aiohttp and make toolbox importable; the current file imports both. Create your own client with a ws:// or wss:// URL and register handlers before connecting. Importing also constructs a module-level demo client without starting it; its demo callback reads a local PDF if that demo client is later connected. Use a new Client instance, as below.

from aiohttp_ws_client import Client

client = Client("wss://example.com/ws")

@client.event()
async def notification(message):
    print("notification:", message.data)

@client.r_event()
async def add(message):
    left = message.data["left"]
    right = message.data["right"]
    return {"total": left + right}

async def connected(active_client):
    response = await active_client.send_req(
        {"left": 2, "right": 3},
        "add",
        timeout=2_000,
    )
    if response is None:
        print("request timed out")
    else:
        print(response.data)

client.on_connect = connected
client.run()

When an event loop already exists

In an async application, await websocket() from your own loop instead of calling the blocking run() helper.

import asyncio
from aiohttp_ws_client import Client

async def main():
    client = Client("wss://example.com/ws")
    await client.websocket(headers={"Authorization": "Bearer …"})

asyncio.run(main())

Events and request/response messages

The dispatcher selects one of three paths after decoding and, when needed, reconstructing a message:

1. Response: an event named RESPONSE with a known response_id completes a pending send_req().
2. Incoming request: any other message with a response_id is sent to a matching r_event handler. Its return value is sent back as RESPONSE.
3. One-way event: a message without response_id is sent to a matching event handler.

One-way event

@client.event()
async def account_updated(message):
    account = message.data
    print(account["id"])

Request handler

@client.r_event()
async def lookup_account(message):
    account_id = message.data["account_id"]
    return {"id": account_id, "status": "active"}

The returned value must be accepted by send_msg(): str, int, float, bytes, dict, or list.

Make a request

reply = await client.send_req(
    {"account_id": 42},
    "lookup_account",
    check_interval=10,  # milliseconds
    timeout=2_000,      # milliseconds
)

if reply is None:
    print("No reply arrived before the deadline")
else:
    print(reply.event, reply.data)

Wire protocol

Every logical message is sent as a JSON text frame with three required keys.

{
  "event": "account_updated",
  "type": "json",
  "data": {"id": 42, "status": "active"}
}

Python value encoding

Other input types raise TypeError. Booleans are not accepted because the implementation checks exact class names rather than using normal numeric subtype behavior.

Chunked messages

_stream_msg() divides a large logical payload into messages with a shared construct ID. The receiver buffers chunks, sorts them by index, and dispatches only after the final chunk arrives.

await client._stream_msg(
    large_payload,
    "archive_uploaded",
    chunk_size=100 * 2**10,
)

Connection lifecycle and reconnects

1. websocket() creates an aiohttp.ClientSession and connects with max_msg_size.
2. It stores the active socket in client.ws, marks the client connected, and schedules on_connect(client).
3. Each incoming frame passes to check(). Handler exceptions are printed and the receive loop continues.
4. When the socket closes, the client waits re_dur seconds and retries up to re_ret times.
5. After retries are exhausted, re_failed becomes true and the connection ends.

A reconnected socket that lasts fewer than 20 seconds is treated as an unstable reconnect and raises TimeoutError into the retry loop.

Ping

ping() uses the application-level PING request handler installed by the constructor. It is not an aiohttp WebSocket control-frame ping.

latency_ms = await client.ping(
    msg_size=128,
    trial_count=5,
    send_as_bytes=False,
    timeout=1_000,
)
print(f"Average: {latency_ms} ms")

Configuration and state

client = Client("wss://example.com/ws")
client.max_msg_size = 16 * 2**20
client.re_ret = 5
client.re_dur = 2
client.re_log = False

Client API reference

The methods below reflect the historical client snapshot. Methods without a leading underscore are presented as public, except where a caveat is explicitly noted.

WSMessage API reference

WSMessage is the decoded object supplied to handlers and returned by successful requests.

Timestamp helpers

ts(dt=None) returns dt.timestamp(), using a naive datetime.utcnow() when omitted. Since timestamp() interprets a naive datetime as local time, the default can be offset on non-UTC hosts; pass an aware UTC datetime for accurate timestamps. from_ts(t) calls datetime.fromtimestamp(t), which returns local time for the host environment.

Client and message objects

A Client owns connection state and callback dictionaries. It creates a native aiohttp.ClientWebSocketResponse in client.ws when websocket() connects. Each received text envelope becomes a separate WSMessage. User handlers receive that message only, so capture the owning client in a closure when sending a reply or another request.

connected describes the receive-loop state; ws can still refer to a closed socket after a disconnect. Check client.ws is not None and not client.ws.closed before low-level operations. reqs and constructs are implementation state, not application storage.

Client workflows

1. Decode a structured message without a connection

Use the class factory to inspect the same object shape an event handler receives. Ordinary messages do not require construct fields.

from aiohttp_ws_client import WSMessage

message = WSMessage.from_msg_dict({
    "event": "RESULT", "type": "json",
    "data": {"count": 3, "ready": True},
    "response_id": "request-1",
})
print(message.event)           # RESULT
print(message.data["count"])   # 3
print(message.response_id)     # request-1

2. Start a request from an event without blocking replies

The server for this example must implement a LOOKUP request event. The event callback schedules work and returns so the receive loop can process the reply. Retain task references until completion and handle request errors inside the task.

import asyncio
from aiohttp_ws_client import Client

client = Client("wss://example.com/ws")
jobs = set()

async def fetch_details(item_id):
    try:
        reply = await client.send_req({"id": item_id}, "LOOKUP", timeout=3000)
        if reply is None:
            print("No details received")
        else:
            print(reply.data)
    except Exception as error:
        print("Lookup failed:", error)

@client.event()
async def ITEM_CHANGED(message):
    task = asyncio.create_task(fetch_details(message.data["id"]))
    jobs.add(task)
    task.add_done_callback(jobs.discard)

client.run()

3. Handle binary payloads and stop deliberately

Bytes passed to send_msg() use base64 in the JSON envelope and arrive decoded. This callback accepts a file fragment, acknowledges its byte count, then closes the current socket while disabling reconnect. It works around the module's broken Client.close(); the separate unclosed-session limitation remains.

from aiohttp_ws_client import Client

client = Client("wss://example.com/ws")
client.max_msg_size = 1024 * 1024

@client.event()
async def FILE_FRAGMENT(message):
    if message.type != "bytes":
        return
    fragment = message.data
    await client.send_msg({"received_bytes": len(fragment)}, "ACK")
    client.re_failed = True
    if client.ws is not None and not client.ws.closed:
        await client.ws.close()

client.run()

Troubleshooting

Known caveats in the current module

• close() checks self.closed even though the constructor does not initialize it, and references ws instead of self.ws. It can raise AttributeError or NameError.
• Reconnects call websocket() without the original headers, so authentication headers from the initial call are not preserved. The initial connection attempt occurs outside the retry loop and may raise directly.
• websocket(headers={}) uses a mutable default argument. The function does not currently mutate it, but callers should still pass an explicit dictionary.
• The ClientSession is not wrapped in async with or explicitly closed, which can produce unclosed-session warnings.
• The receive loop awaits event handlers directly. If such a handler awaits send_req() on the same client, the loop cannot read the reply until that handler returns. Schedule outbound request work in a separate task, as the built-in on_connect path does.
• send_req() polls shared state instead of awaiting a future; very small or invalid check_interval values can create division errors or ineffective timeouts.
• ping(interval=.5) passes .5 as milliseconds, which truncates timeout / check_interval and sleeps for 0.0005 seconds per poll.
• send_msg() prints every serialized outgoing message, which may expose payloads in logs.
• _stream_msg() and construct_msg() disagree on some text/byte joins in the current module. Validate chunked messages end to end.
• WSMessage.from_msg_dict() assigns construct_id from response_id instead of construct_id.
• The generated reference previously listed a module-level on_connect(client) function. That function is demo code that reads a local PDF and is not part of the reusable client API.
• The client accepts an unusually large default message size (about 3.9 GiB). Set a realistic limit to reduce memory-exhaustion risk.