SocketSpec

FastAPI-style WebSocket framework with built-in interactive docs and testing.

Install

pip install socketspec[fastapi]

Quickstart

from fastapi import FastAPI
from pydantic import BaseModel
from socketspec import SocketApp
from socketspec.adapters.fastapi import mount

socket = SocketApp(docs=True)

class ChatMessage(BaseModel):
    room: str
    text: str

class MessageAck(BaseModel):
    status: str
    message_id: str

@socket.on(
    "send_message",
    description="Send a chat message to a room.",
    tags=["chat"],
    emits=[Emits("message_ack", model=MessageAck, description="Delivery confirmation")],
    broadcasts=[Broadcasts("new_message", room="chat:{room}", description="Delivered to room members")],
)
async def send_message(conn, payload: ChatMessage) -> None:
    await conn.emit("message_ack", {"status": "ok", "message_id": "abc123"})
    await socket.rooms.broadcast(
        "chat:" + payload.room,
        "new_message",
        {"from": conn.id, "text": payload.text},
    )

app = FastAPI()
mount(socket, app, path="/ws")

uvicorn main:app --reload

Docs UI

Open /socket-docs in your browser after starting the server.

• Click an event card to expand its schema
• Hit Try it out to send a live WebSocket message
• See the server response appear inline, without leaving the browser
• Open a second tab and connect as a different user to test broadcasts

Why SocketSpec

Core Concepts

Event handlers look exactly like FastAPI route handlers:

@socket.on("join_room", tags=["rooms"])
async def join_room(conn: Connection, payload: JoinPayload) -> None:
    await socket.rooms.join(conn, f"chat:{payload.room_id}")

Rooms with pattern-based guards:

@socket.room_guard("admin:{room}")
async def admin_only(conn: Connection, room: str) -> bool:
    return conn.identity.role == "admin"

Testing without a real server:

from socketspec.testing import TestClient

async def test_send_message():
    async with TestClient(socket) as client:
        conn = await client.connect()
        await conn.send("send_message", {"room": "general", "text": "hello"})
        response = await conn.receive()
        assert response["event"] == "message_ack"

Links

• Documentation
• GitHub
• PyPI
• Changelog
• Contributing

License

Apache 2.0 — see LICENSE. Created and maintained by Laiba Shahab.