SDKs
Python

Python SDK

Async-first client for Python 3.10+.

Installation

pip install konet

Quick Start

import asyncio
from konet import KonetClient
 
async def main():
    async with KonetClient("ws://localhost:4000/socket", token="<anon_key>") as client:
        channel = client.channel("room:lobby")
        await channel.subscribe()
 
        channel.on("message", lambda payload: print("Received:", payload))
        await channel.send("message", {"text": "Hello from Python!"})
 
        await asyncio.sleep(60)
 
asyncio.run(main())

async with connects on enter and disconnects on exit. If you prefer to manage the lifecycle yourself, call connect() / disconnect() explicitly:

client = KonetClient("ws://localhost:4000/socket", token="<anon_key>")
await client.connect()
# ...
await client.disconnect()

API

# token is keyword-only
client = KonetClient(
    url,
    token=...,
    heartbeat_interval=30.0,
    reconnect_delay=1.0,
    max_reconnect_tries=10,
)
await client.connect()
await client.disconnect()
 
channel = client.channel("room:lobby")
await channel.subscribe()
off = channel.on(event, callback)   # off() removes the handler
await channel.send(event, payload)
await channel.unsubscribe()

Testing

pip install pytest pytest-asyncio
pytest

Binary Frames & Floor Control

For anything at a media rate. Take the floor before sending — a binary frame without it is refused. See the protocol page for the guarantees.

channel = client.channel("room:team-42:ptt")
await channel.subscribe()
 
def on_frame(data: bytes) -> None:
    ...  # an incoming frame
 
channel.on_binary("a", on_frame)
 
try:
    await channel.acquire_floor()
    await channel.send_binary("a", frame)
except RuntimeError as error:
    ...  # someone else is talking; the error names them
finally:
    await channel.release_floor()

send_binary copies its argument, so a caller may reuse one buffer for every frame.

GoAPI Reference