Define server methods effortlessly and protect them with custom rules (predicates).

class MyHandler(webshocket.WebSocketHandler):
    @webshocket.rpc_method(alias_name="add")
    async def add(self, _: webshocket.ClientConnection, a: int, b: int):
        return a + b

    # Unique: Use built-in predicates for clean access control
    @webshocket.rpc_method(requires=webshocket.Is("is_admin"))
    async def secret_function(self, conn: webshocket.ClientConnection):
        return "Sensitive Data"

No more look-up tables. Assign data directly to the client connection; Webshocket handles the persistence for you.

    @webshocket.rpc_method()
    async def login(self, connection: webshocket.ClientConnection, user_id: str):
        # Direct attribute assignment persists for the session
        connection.user_id = user_id
        connection.is_admin = True

        # Subscribe to updates immediately
        connection.subscribe("broadcast-channel")

Protect your RPC methods from abuse with a simple decorator. Supports human-readable periods and optional auto-disconnect.

class MyHandler(webshocket.WebSocketHandler):
    @webshocket.rate_limit(limit=5, period="1m")  # 5 calls per minute
    @webshocket.rpc_method()
    async def expensive_operation(self, connection: webshocket.ClientConnection, query: str):
        return await run_ai_model(query)

    @webshocket.rate_limit(limit=100, period="10s", disconnect_on_limit_exceeded=True)
    @webshocket.rpc_method()
    async def chat_message(self, connection: webshocket.ClientConnection, msg: str):
        return await process_message(msg)

Webshocket supports powerful wildcard subscriptions using glob patterns (*, ?, []). This allows clients to subscribe to a broad category of updates with a single call.

class StockHandler(webshocket.WebSocketHandler):
    @webshocket.rpc_method()
    async def watch_sector(self, connection: webshocket.ClientConnection, sector: str):
        # Client subscribes to a pattern like "tech.*"
        connection.subscribe(f"{sector}.*")

    def broadcast_stock_price(self, symbol: str, price: float):
        # If symbol is "tech.AAPL", it automatically routes to anyone
        # subscribed to "tech.*" OR "tech.AAPL"
        self.publish(symbol, {"price": price})

Smart Filtering (Predicates)

You can also combine subscriptions with Predicates to filter recipients dynamically without creating thousands of sub-channels.

    @webshocket.rpc_method()
    async def join_room(self, connection: webshocket.ClientConnection, room: str, team: str):
        connection.subscribe(room)
        connection.team = team # Persistent session state

    def send_team_update(self, room: str, team: str, data: dict):
        # Only publish to clients in 'room' whose 'team' attribute equals 'team'
        self.publish(room, data, predicate=webshocket.IsEqual("team", team))

The client handles reconnection automatically, no manual retry loops needed.

async def main():
    client = webshocket.WebSocketClient("ws://your-tunnel.url")
    await client.connect(retry=True, max_retry_attempt=5, retry_interval=2)

    result = await client.send_rpc("add", 10, 20)
    print(result.data)  # 30

Designed to run perfectly behind free HTTP tunnels, making it the easiest way to expose a local AI or IoT project to the world.

async def main():
    server = webshocket.WebSocketServer("0.0.0.0", 5000, clientHandler=MyHandler)
    async with server:
        await server.serve_forever()

Webshocket is designed to be language-agnostic. While the Python client is optimized with msgspec, the server natively understands standard JSON packets. This means you can build a client in JavaScript, Java, or C# using nothing but the standard library.

// Example: Standard Browser JavaScript Client
const socket = new WebSocket("ws://your-tunnel.url");

socket.onopen = () => {
	const rpcRequest = {
		rpc: {
			type: "request",
			method: "add",
			args: [10, 20],
			kwargs: {},
		},
		source: 5,
	};

	socket.send(JSON.stringify(rpcRequest));
};

socket.onmessage = async (event) => {
	const packet = JSON.parse(await event.data.text());
	console.log("Result:", packet.rpc.response); // 30
};

Stream data from server to client in real-time using async generators. The client iterates over chunks as they arrive. This is perfect for AI token streaming, live feeds, or large result sets.

# Server - yield chunks as they're produced
class MyHandler(webshocket.WebSocketHandler):
    @webshocket.rpc_method()
    async def generate_report(self, connection: webshocket.ClientConnection, query: str):
        async for chunk in run_ai_model(query):
            yield chunk  # Each yield sends a packet to the client

# Client - iterate over the stream
async for packet in client.stream_rpc("generate_report", "Summarize Q4 earnings"):
    print(packet.data, end="", flush=True)