Composable, reusable WebSocket components for any ASGI framework (Django, FastAPI, Litestar)
Hi all, I'm the maintainer of a small channels (WebSocket) extension library for Django (and FastAPI too). While using and maintaining it, I started thinking it could become a small framework as well: composable and framework-independent, so it could be reused across Django/FastAPI/Litestar/... as long as the framework supports ASGI. Before going further, I'm putting the blueprint out here to compare notes with people who work with WebSockets regularly. If you have ever worked with WebSockets, I hope you can share any ideas, info, pain points, or suggestions you have.
Prerequisites, what my library already has:
- Function-like handlers rather than
while True+ if/else - Automatic AsyncAPI doc generation
- Full type hints
- A testing kit
- Support for all ASGI-based frameworks (Django, FastAPI, ...)
At a glance, it looks like this:
@ws_handler(output_type=ChatNotificationMessage)
async def handle_chat(self, message: ChatMessage) -> None:
# Automatically routed, validated, and type-safe
await self.broadcast_message(
ChatNotificationMessage(payload=message.payload)
)
@ws_handler
async def handle_ping(self, message: PingMessage) -> PongMessage:
return PongMessage() # Auto-documented in AsyncAPI
If you have ever worked with WebSockets, I think you get the idea of what it does here.
Recently I added the Topic feature, which is composable and reusable. It came out of a multiplexing feature request, and I was inspired by Phoenix Channels. It looks something like this:
class DiscussionTopic(Topic):
pattern = "discussion:{pk}"
async def authorize(self, pk: str) -> bool:
return await user_can_view(self.scope["user"], pk)
@ws_handler
async def handle_reply(self, message: ReplyMessage) -> ReplyCreatedMessage:
return ReplyCreatedMessage(payload=message.payload)
@event_handler
async def handle_new_reply(self, event: NewReplyEvent) -> ReplyCreatedMessage:
return ReplyCreatedMessage(payload=event.payload)
And you use it like this:
class HubConsumer(AsyncJsonWebsocketConsumer):
authenticator_class = JWTAuthenticator
topics = [DiscussionTopic, RoomTopic]
In short, topics let you multiplex: subscribe, publish messages, unsubscribe, and so on, all over the same socket. So you can reuse a single WebSocket connection and just add or compose multiple topics, i.e. multiple WebSocket handlers.
That made me think: if we could create reusable topics such as Notification, Streaming, Voice, AI Agent, and so on, which users could easily install or copy and then modify or inherit from in a structured way, WebSocket handling would become much more structured and easier. The idea is similar to DRF and its ecosystem, and the composable/reusable part would work like shadcn: copy it, own it, and modify the code freely.
What would you use it for? As I mentioned above: notifications, streaming, voice, AI agents, and so on. I have done a lot of WebSocket work, and I keep having to redefine the same things over and over. There is no reusable approach like the ones we have for REST APIs. Another example is using Pydantic AI with the AG-UI protocol but over WebSockets, defined in a reusable way.
So, if you already know of an existing open source solution or library similar to this idea, it would be great if you could share it here. And if this resonates with you, a comment would help, both to add more insight and to give some encouragement to actually build this.