API Reference

This page documents the Python client library API.

Module: `roomz`

The main module exports the primary client classes and exceptions.

from roomz import AsyncClient, SyncClient
from roomz import ClientError, ConnectionError, AuthenticationError
from roomz import ConnectionState

AsyncClient

Primary async WebSocket client for Roomz real-time chat.

Constructor

AsyncClient(
  server_url: str,
  session_token: str = "",
  *,
  reconnect: bool = True,
  reconnect_delay: float = 1.0,
  max_reconnect_attempts: int = 5,
  connection_timeout: float = 10.0,
  session_cache_file: str | Path | None = None,
)

Parameters:

Parameter	Type	Default	Description
`server_url`	`str`	required	WebSocket server URL (e.g., `http://localhost:8000`)
`session_token`	`str`	`""`	Magic link token for authentication
`reconnect`	`bool`	`True`	Enable automatic reconnection
`reconnect_delay`	`float`	`1.0`	Initial reconnection delay in seconds
`max_reconnect_attempts`	`int`	`5`	Maximum reconnection attempts
`connection_timeout`	`float`	`10.0`	Connection timeout in seconds
`session_cache_file`	`str \| Path \| None`	`None`	Path to cache session cookie

Properties

`connected` (property)

@property
def connected() -> bool

Returns True if WebSocket is connected and authenticated.

`user` (property)

@property
def user() -> dict | None

Returns current user info or None if not authenticated.

Returns: {"id": str, "email": str, "display_name": str | None} or None

`connection_state` (property)

@property
def connection_state() -> ConnectionState

Returns current connection state.

Returns: ConnectionState enum value

Methods

`on(event, handler)`

def on(event: str, handler: EventHandler) -> None

Register an event handler.

Parameters:

Parameter	Type	Description
`event`	`str`	Event name: `authenticated`, `message`, `user_joined`, `user_left`, `disconnect`, `error`
`handler`	`Callable`	Callback function receiving event data (sync or async)

Example:

def on_message(data):
    print(f"{data['user']['email']}: {data['content']}")

client.on("message", on_message)

`off(event, handler=None)`

def off(event: str, handler: EventHandler | None = None) -> None

Unregister an event handler.

Parameters:

Parameter	Type	Description
`event`	`str`	Event name
`handler`	`Callable \| None`	Specific handler to remove, or `None` to remove all handlers for event

`login(email)`

async def login(email: str) -> bool

Request a magic link for authentication.

Parameters:

Parameter	Type	Description
`email`	`str`	User email address

Returns: True if magic link was requested successfully

Raises: ConnectionError if request fails

Example:

success = await client.login("user@example.com")
# Check email for magic link...

`connect(session_token=None)`

async def connect(session_token: str | None = None) -> None

Establish WebSocket connection with authentication.

Parameters:

Parameter	Type	Description
`session_token`	`str \| None`	Magic link token. If not provided, uses cached session.

Raises:

ConnectionError if connection fails
AuthenticationError if authentication fails

Example:

# Connect with token
await client.connect(token="magic-link-token")

# Or reconnect with cached session
await client.connect()

`disconnect()`

async def disconnect() -> None

Close WebSocket connection.

Example:

await client.disconnect()

`send(content)`

async def send(content: str) -> dict

Send a chat message.

Parameters:

Parameter	Type	Description
`content`	`str`	Message content

Returns:

On success:

{
  "status": "ok",
  "message_id": "uuid",
  "timestamp": "2024-01-01T12:00:00Z"
}

On failure:

{
  "error": "Error message",
  "code": 408  # HTTP status code
}

Raises: ConnectionError if not connected

Example:

result = await client.send("Hello, world!")
if "error" in result:
    print(f"Failed: {result['error']}")
else:
    print(f"Message ID: {result['message_id']}")

`clear_cached_session()`

def clear_cached_session() -> None

Clear cached session cookie.

Example:

client.clear_cached_session()

`set_display_name(name)`

async def set_display_name(name: str | None) -> dict

Set or clear the display name for the current user.

Parameters:

Parameter	Type	Description
`name`	`str \| None`	Display name to set, or `None` to clear

Returns:

On success:

{
  "status": "ok",
  "display_name": "Alice"  # or null if cleared
}

On failure:

{
  "error": "Error message",
  "code": 400
}

Raises: ConnectionError if not connected

Example:

# Set display name
result = await client.set_display_name("Alice")
print(f"Display name set to: {result['display_name']}")

# Clear display name
result = await client.set_display_name(None)
print("Display name cleared")

Note: Display names appear as "Display Name (email)" in chat. When cleared, only the email is shown.

Context Manager

AsyncClient supports async context manager:

async with AsyncClient(server_url, session_token) as client:
    client.on("message", handler)
    await client.send("Hello!")

SyncClient

Synchronous wrapper around AsyncClient with background thread.

Constructor

SyncClient(
  server_url: str,
  session_token: str,
  *,
  reconnect: bool = True,
  reconnect_delay: float = 1.0,
  max_reconnect_attempts: int = 5,
  connection_timeout: float = 10.0,
)

Note: session_cache_file is not supported in SyncClient.

Methods

All methods are synchronous equivalents of AsyncClient:

connect() - Establish connection
disconnect() - Close connection
send(content) - Send message (synchronous)
on(event, handler) - Register event handler
off(event, handler=None) - Unregister event handler

Context Manager

with SyncClient(server_url, session_token) as client:
    client.on("message", handler)
    result = client.send("Hello!")

ConnectionState Enum

class ConnectionState(Enum):
  DISCONNECTED = "disconnected"
  CONNECTING = "connecting"
  CONNECTED = "connected"
  RECONNECTING = "reconnecting"

Exceptions

ClientError

Base exception for all client errors.

class ClientError(Exception):
    """Base exception for client errors."""

ConnectionError

Raised when connection fails.

class ConnectionError(ClientError):
    """Connection-related errors."""

AuthenticationError

Raised when authentication fails.

class AuthenticationError(ClientError):
    """Authentication-related errors."""

Events

`authenticated`

Emitted when successfully authenticated.

Data:

{
  "user": {"id": "user:email@example.com", "email": "email@example.com", "display_name": null},
  "channel": "user:email@example.com",
  "server_time": "2024-01-01T12:00:00Z"
}

`message`

Emitted when a message is received.

Data:

{
  "id": "uuid",
  "user": {"id": "user:email@example.com", "email": "email@example.com", "display_name": "Alice"},
  "content": "Message text",
  "timestamp": "2024-01-01T12:00:00Z"
}

`user_joined`

Emitted when a user joins the channel.

Data:

{
  "user": {"id": "user:email@example.com", "email": "email@example.com", "display_name": "Alice"},
  "timestamp": "2024-01-01T12:00:00Z",
  "total_connections": 1
}

`user_left`

Emitted when a user leaves the channel.

Data:

{
  "user": {"id": "user:email@example.com", "email": "email@example.com", "display_name": "Alice"},
  "timestamp": "2024-01-01T12:00:00Z",
  "remaining_connections": 0
}

`display_name_changed`

Emitted when a user changes their display name.

Data:

{
  "user": {"id": "user:email@example.com", "email": "email@example.com"},
  "display_name": "Alice",
  "timestamp": "2024-01-01T12:00:00Z"
}

`disconnect`

Emitted when disconnected from server.

Data: {}

`error`

Emitted when an error occurs.

Data:

{
  "error": "Error message",
  "code": 500
}

CLI Commands

The roomz-cli command provides a terminal interface.

`roomz-cli`

roomz-cli [--server URL]

Options:

Option	Default	Description
`--server`, `-s`	`http://localhost:8000`	Server URL

In-CLI Commands

Command	Description
`/login <email>`	Request magic link
`/token <token>`	Connect with token
`/name <name>`	Set display name (shown as “Name (email)”)
`/name`	Clear display name (show email only)
`/logout`	Disconnect and clear session
`/quit`	Exit the CLI

API Reference

Module: roomz

AsyncClient

Constructor

Properties

connected (property)

user (property)

connection_state (property)

Methods

on(event, handler)

off(event, handler=None)

login(email)

connect(session_token=None)

disconnect()

send(content)

clear_cached_session()

set_display_name(name)

Context Manager

SyncClient

Constructor

Methods

Context Manager

ConnectionState Enum

Exceptions

ClientError

ConnectionError

AuthenticationError

Events

authenticated

message

user_joined

user_left

display_name_changed

disconnect

error

CLI Commands

roomz-cli

In-CLI Commands

Module: `roomz`

`connected` (property)

`user` (property)

`connection_state` (property)

`on(event, handler)`

`off(event, handler=None)`

`login(email)`

`connect(session_token=None)`

`disconnect()`

`send(content)`

`clear_cached_session()`

`set_display_name(name)`

`authenticated`

`message`

`user_joined`

`user_left`

`display_name_changed`

`disconnect`

`error`

`roomz-cli`