Standalone aiohttp Server

SignalRServer.serve() runs a fully self-contained HTTP + WebSocket server built on aiohttp. No external ASGI server is required.

Basic usage

import asyncio

from aiosignalr.server import Hub, SignalRServer


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


async def main() -> None:
    server = SignalRServer(ChatHub)
    host, port = await server.serve("127.0.0.1", 8080, path="/hub")
    print(f"listening on http://{host}:{port}/hub")


asyncio.run(main())

What serve() does

1. Builds an aiohttp application with routes for {path}/negotiate and {path} (WebSocket upgrade, SSE, Long Polling, HTTP POST).
2. Binds a TCP socket. Passing port=0 lets the OS pick a free port; the bound (host, port) is returned.
3. Runs until you close it.

The underlying aiohttp application

create_application() returns the aiohttp web.Application directly, so you can customize middleware or mount it into a larger aiohttp app:

app = server.create_application(path="/hub")
# app.middlewares.append(...)

Endpoint behavior

Path / method	Purpose
POST {path}/negotiate	Negotiation (with negotiateVersion and useStatefulReconnect support)
GET {path} (WebSocket upgrade)	Full-duplex WebSocket transport
GET {path} (Accept: text/event-stream)	Server-Sent Events transport
GET {path}	Long Polling transport
POST {path}	Client-to-server HTTP POST
DELETE {path}	Graceful connection teardown

Shutting down

await server.close()

close() cleans up the aiohttp runner. It is safe to call from a signal handler.

Configuration

All knobs live in ServerOptions:

server = SignalRServer(
    ChatHub,
    options=ServerOptions(
        keep_alive_interval=10.0,
        client_timeout_interval=20.0,
        enable_detailed_errors=True,
        allow_stateful_reconnects=True,
    ),
)