TuoniWebSocket¶

class tuoni.TuoniWebSocket(c2)[source]¶

Real-time WebSocket client for the Tuoni server using SockJS/STOMP.

Subscribes to server-pushed events (agent checkins, command results, listener changes, etc.) without polling. Requires the websocket-client package.

TOPIC_AGENT¶

New agent registered.

Type:

str

TOPIC_AGENT_HEARTBEAT¶

Agent heartbeat/checkin.

Type:

str

TOPIC_AGENT_ACTIVE¶

Agent active-status change.

Type:

str

TOPIC_AGENT_METADATA¶

Agent metadata changed.

Type:

str

TOPIC_COMMAND_SENT¶

Command dispatched to agent.

Type:

str

TOPIC_COMMAND_RESULT¶

Command result received.

Type:

str

TOPIC_COMMAND_TEMPLATE_NEW¶

New command template available.

Type:

str

TOPIC_COMMAND_TEMPLATE_REMOVED¶

Command template removed.

Type:

str

TOPIC_LISTENER_NEW¶

New listener created.

Type:

str

TOPIC_LISTENER_REMOVED¶

Listener deleted.

Type:

str

TOPIC_LISTENER_MODIFIED¶

Listener configuration changed.

Type:

str

TOPIC_PAYLOAD_NEW¶

New payload created.

Type:

str

TOPIC_PAYLOAD_ARCHIVED¶

Payload archived.

Type:

str

TOPIC_USER_NEW¶

New user created.

Type:

str

TOPIC_USER_MODIFIED¶

User modified.

Type:

str

Examples

>>> ws = tuoni_server.connect_websocket()
>>> results = []
>>> ws.on_command_result(lambda data: results.append(data))
>>> agent.send_command(TuoniCommandLs(".", 1))
>>> # callback fires when agent returns the result

connect(timeout: float = 10) → bool[source]¶

Connect to the server’s WebSocket endpoint and complete the STOMP handshake.

Parameters:

timeout (float) – Seconds to wait for the STOMP CONNECTED frame.

Returns:

True if the STOMP session was established within timeout seconds.

Return type:

bool

Examples

>>> ws = TuoniWebSocket(tuoni_server)
>>> if not ws.connect():
>>>     print("WebSocket connection failed")

disconnect()[source]¶

Send STOMP DISCONNECT and close the WebSocket.

property is_connected: bool¶

True if the STOMP session is active.

on_agent_active_change(callback)[source]¶

Register a callback for agent active-status changes.

on_agent_heartbeat(callback)[source]¶

Register a callback for agent heartbeats.

on_command_result(callback)[source]¶

Register a callback for command results.

on_command_sent(callback)[source]¶

Register a callback for commands dispatched to agents.

on_event(topic: str, callback)[source]¶

Subscribe to a STOMP topic and register a callback.

If already connected the subscription is sent immediately; otherwise it is queued and sent once the STOMP session is established.

Parameters:

• topic (str) – STOMP topic (e.g. TuoniWebSocket.TOPIC_COMMAND_RESULT).

• callback (callable) – Called with the deserialized JSON payload when a message arrives. Runs in a daemon thread.

Examples

>>> ws.on_event(TuoniWebSocket.TOPIC_AGENT, lambda data: print("new agent:", data))

on_listener_modified(callback)[source]¶

Register a callback for listener modifications.

on_listener_new(callback)[source]¶

Register a callback for new listeners.

on_new_agent(callback)[source]¶

Register a callback for new agent registrations.

on_payload_new(callback)[source]¶

Register a callback for new payloads.