Chat Server Example

This is the runnable example shipped in examples/chat_server.py. It demonstrates lifecycle hooks, groups, and broadcasting.

Server

import asyncio

from aiosignalr.server import Hub, SignalRServer


class ChatHub(Hub):
    """Broadcasts messages to every connected client."""

    async def on_connected(self) -> None:
        await self.clients.all_.send("system", f"{self.context.connection_id} joined")
        await self.groups.add("general")

    async def on_disconnected(self, exception: BaseException | None) -> None:
        await self.clients.all_.send("system", f"{self.context.connection_id} left")

    async def send_message(self, text: str) -> None:
        await self.clients.group("general").send("message", self.context.connection_id, text)

    async def whisper(self, connection_id: str, text: str) -> None:
        await self.clients.client(connection_id).send("message", self.context.connection_id, text)


async def main() -> None:
    server = SignalRServer(ChatHub)
    host, port = await server.serve("127.0.0.1", 8080, path="/hub")
    print(f"Chat hub listening on http://{host}:{port}/hub")
    try:
        await asyncio.Event().wait()
    except KeyboardInterrupt:
        pass
    finally:
        await server.close()


if __name__ == "__main__":
    asyncio.run(main())

Run it:

uv run python examples/chat_server.py
# Chat hub listening on http://127.0.0.1:8080/hub

Client

import asyncio

from aiosignalr.client import HubConnection


async def main() -> None:
    conn = HubConnection()
    conn.on("system", lambda text: print("[system]", text))
    conn.on("message", lambda sender, text: print(f"[{sender}] {text}"))

    await conn.start("http://127.0.0.1:8080/hub")

    await conn.send("send_message", "hello everyone")

    # Keep the client alive and react to messages.
    await asyncio.sleep(5)
    await conn.stop()


asyncio.run(main())

What this exercises

• on_connected / on_disconnected lifecycle hooks.
• Group membership (groups.add) and group broadcast (clients.group("general").send).
• Broadcast to everyone (clients.all_.send).
• Targeted sends via a connection id (clients.client(connection_id).send).