pysignalr

import asyncio from contextlib import suppress from typing import Any, Dict, List from pysignalr.client import SignalRClient from pysignalr.messages import CompletionMessage import os async def on_open() -> None: print('Connected to the server') async def on_close() -> None: print('Disconnected from the server') async def on_message(message: List[Dict[str, Any]]) -> None: print(f'Received message: {message}') async def on_client_result(message: list[dict[str, Any]]) -> str: print(f'Received message requesting result: {message}') return 'reply_from_client' async def on_error(message: CompletionMessage) -> None: print(f'Received error: {message.error}') async def main() -> None: # Replace with your SignalR hub URL # For example, a public API like TzKT.io or your own local/remote hub signalr_url = os.environ.get('SIGNALR_HUB_URL', 'https://api.tzkt.io/v1/ws') access_token = os.environ.get('SIGNALR_ACCESS_TOKEN', '') # Optional: for authenticated hubs client_args = {'url': signalr_url} if access_token: # For authenticated hubs, provide an access token factory client_args['access_token_factory'] = lambda: access_token client = SignalRClient(**client_args) client.on_open(on_open) client.on_close(on_close) client.on_error(on_error) # Register handlers for specific events from the server client.on('operations', on_message) # Example: subscribing to 'operations' event client.on('client_result', on_client_result) # Example: handling a server request for a client result await asyncio.gather( client.run(), # Example: Sending a message to the server (e.g., to subscribe to a topic) client.send('SubscribeToOperations', [{}]), ) if __name__ == '__main__': with suppress(KeyboardInterrupt, asyncio.CancelledError): asyncio.run(main())