Skip to content

websockets

A WebSocket carries text, so use request_json and parse_json. This uses the websockets.asyncio client from websockets 13 and later:

import asyncio
import logging

from websockets.asyncio.client import connect

from jsonrpcclient import Error, Ok, parse_json, request_json


async def main() -> None:
    async with connect("ws://localhost:8000") as websocket:
        await websocket.send(request_json("ping"))
        parsed = parse_json(await websocket.recv())
    if isinstance(parsed, Ok):
        print(parsed.result)
    elif isinstance(parsed, Error):
        logging.error(parsed.message)


asyncio.run(main())

A WebSocket connection can carry many requests at once, and the server may answer them in any order. If you send several before reading, match the replies to your requests by id.