aiosignalr

High-performance, asyncio-native implementation of the SignalR hub protocol for Python. Provides both a client and a server, covering the WebSocket, Server-Sent Events, Long Polling and HTTP Post transports with JSON and MessagePack message encoding.

Documentation

Full documentation (English & 简体中文) is available at:

https://truerou.github.io/aiosignalr/

• Getting Started
• Client Guide
• Server Guide
• Stateful Reconnect

The site source lives in website/ (Docusaurus) and is deployed to GitHub Pages by the deploy-docs workflow.

Features

• Pure asyncio, no blocking calls.
• Full-duplex client and server over WebSocket.
• Transport fallback: the client negotiates transports in server-declared order (WebSocket → Server-Sent Events → Long Polling).
• Both the JSON (json, 0x1E-delimited) and MessagePack (messagepack, VarInt length-prefixed) protocols.
• Client: invoke, stream, send, client-to-server upload streams, server method handlers (on), keep-alive pings, server-timeout detection, automatic reconnect with a pluggable retry policy.
• Server: hub methods (single result and streaming), groups, broadcast to all/connection/group/user, client results, cancellation, upload-stream parameters, negotiation and handshake protocol selection.
• Server hosting: standalone asyncio server or an ASGI application mountable in uvicorn/FastAPI.
• Stateful reconnect on both sides: useAck / Ack / Sequence message buffering that preserves in-flight invocations and broadcasts across a WebSocket drop, with exactly-once delivery (interoperates with the ASP.NET Core implementation).
• Interop test-suite: aiosignalr ↔ real ASP.NET Core SignalR server, and a real ASP.NET Core SignalR client ↔ aiosignalr server (see tests/interop/).

Installation

uv add aiosignalr        # or: pip install aiosignalr

Requires Python 3.11+.

Client

import asyncio
from aiosignalr.client import HubConnection


async def main() -> None:
    connection = HubConnection()
    connection.on("message", lambda text: print("got:", text))

    await connection.start("ws://127.0.0.1:8080/hub")
    result = await connection.invoke("Add", 40, 2)
    print("Add(40, 2) =", result)

    async for item in await connection.stream("Counter", 3):
        print("stream item:", item)

    await connection.send("Notify", "hello")
    await connection.stop()


asyncio.run(main())

Server

import asyncio
from aiosignalr.server import Hub, ServerOptions, SignalRServer


class ChatHub(Hub):
    async def on_connected(self) -> None:
        await self.clients.all_.send("message", f"User {self.context.connection_id} joined")

    async def echo(self, text: str) -> str:
        return text

    async def counter(self, n: int):
        for i in range(n):
            yield i
            await asyncio.sleep(0.01)


async def main() -> None:
    server = SignalRServer(
        ChatHub,
        options=ServerOptions(allow_stateful_reconnects=True),
    )
    await server.serve("127.0.0.1", 8080, path="/hub")


asyncio.run(main())

To mount the server inside an existing ASGI application (uvicorn/FastAPI):

app = server.asgi_app()  # pass to uvicorn.run(app, ...)

Stateful reconnect

Both the client and the server implement the ASP.NET Core stateful-reconnect ("ack") protocol. When enabled, unacknowledged messages are buffered and replayed after a transport drop so in-flight invocations, streams and broadcasts survive without loss or duplication.

Client:

connection = HubConnection().with_stateful_reconnect()

Server:

SignalRServer(ChatHub, options=ServerOptions(allow_stateful_reconnects=True))

Stateful reconnect requires the WebSockets transport. The client sends a HubProtocol version-2 handshake and the negotiate response advertises useStatefulReconnect when the server agrees; the two sides then exchange Ack/Sequence messages and replay buffered messages across the reconnect.

Tests

uv run ruff format .
uv run ruff check .
uv run mypy src
uv run pytest                 # self-tests
uv run pytest tests/interop   # requires dotnet (real ASP.NET Core interop)

Documentation site

The docs site lives in website/ (Docusaurus, English + 简体中文) and is deployed automatically to GitHub Pages by .github/workflows/deploy-docs.yml on every push touching website/**.

Local development

cd website
npm install
npm start          # dev server at http://localhost:3000/aiosignalr/
npm run build      # static build into website/build/

One-time GitHub Pages setup

The workflow uses the official actions/deploy-pages action, which requires the Pages source to be GitHub Actions:

1. Open Settings → Pages.
2. Under Build and deployment → Source, select GitHub Actions.
3. Push the workflow to main. The deploy job publishes https://truerou.github.io/aiosignalr/ (auto-enabled after the first run).

Adding a language

• Write the docs under website/docs/ (English is the default locale).
• Add the locale to website/docusaurus.config.js → i18n.locales.
• Create website/i18n/<locale>/docusaurus-plugin-content-docs/current/ and mirror the file tree.
• Run cd website && npm run write-translations -- --locale <locale> to scaffold the theme/code translations, then translate the JSON files.

License

MIT