Quick start

Expose methods at /ws/, connect once, then call in either direction. Choose your framework below. Installation, method and server tabs stay in sync; the Python and browser clients work with every server shown here.

Install

Use Python 3.10+ in a virtual environment. Install the package for your server:

aiohttp

pip install wsrpc-aiohttp

FastAPI

pip install 'wsrpc-asgi[client,fastapi]' uvicorn

Starlette

pip install 'wsrpc-asgi[client,starlette]' uvicorn

Django

pip install 'wsrpc-asgi[client]' django uvicorn

Channels

pip install 'wsrpc-asgi[client]' django channels uvicorn

Quart

pip install 'wsrpc-asgi[client]' quart uvicorn

Litestar

pip install 'wsrpc-asgi[client]' litestar uvicorn

Falcon

pip install 'wsrpc-asgi[client]' falcon uvicorn

Sanic

pip install 'wsrpc-asgi[client]' sanic uvicorn

BlackSheep

pip install 'wsrpc-asgi[client]' blacksheep uvicorn

Bare ASGI

pip install 'wsrpc-asgi[client]' uvicorn

For a source installation of the ASGI library, use the same extras with its Git repository instead of the distribution name, for example:

pip install 'wsrpc-asgi[client,fastapi] @ git+https://github.com/wsrpc/wsrpc-asgi.git' uvicorn

Define the RPC methods

Every server below exposes the same two methods: echo returns its keyword arguments, and callback asks the client to double a value.

The method definitions below follow your framework selection.

aiohttp

These methods are included in the complete app.py server below.

from wsrpc_aiohttp import WebSocketAsync
class RPC(WebSocketAsync):
    pass


async def echo(socket, **params):
    return params


async def callback(socket, *, value):
    return await socket.proxy.double(value=value)


RPC.add_route("echo", echo)
RPC.add_route("callback", callback)

FastAPI

These methods are included in the complete app.py server below.

from wsrpc_asgi import WebSocketAsync
class RPC(WebSocketAsync):
    pass


async def echo(socket, **params):
    return params


async def callback(socket, *, value):
    return await socket.proxy.double(value=value)


RPC.add_route("echo", echo)
RPC.add_route("callback", callback)

Starlette

These methods are included in the complete app.py server below.

from wsrpc_asgi import WebSocketAsync
class RPC(WebSocketAsync):
    pass


async def echo(socket, **params):
    return params


async def callback(socket, *, value):
    return await socket.proxy.double(value=value)


RPC.add_route("echo", echo)
RPC.add_route("callback", callback)

Django

These methods are included in the complete app.py server below.

from wsrpc_asgi import WebSocketAsync
class RPC(WebSocketAsync):
    pass


async def echo(socket, **params):
    return params


async def callback(socket, *, value):
    return await socket.proxy.double(value=value)


RPC.add_route("echo", echo)
RPC.add_route("callback", callback)

Channels

These methods are included in the complete app.py server below.

from wsrpc_asgi import WebSocketAsync
class RPC(WebSocketAsync):
    pass


async def echo(socket, **params):
    return params


async def callback(socket, *, value):
    return await socket.proxy.double(value=value)


RPC.add_route("echo", echo)
RPC.add_route("callback", callback)

Quart

These methods are included in the complete app.py server below.

from wsrpc_asgi import WebSocketAsync
class RPC(WebSocketAsync):
    pass


async def echo(socket, **params):
    return params


async def callback(socket, *, value):
    return await socket.proxy.double(value=value)


RPC.add_route("echo", echo)
RPC.add_route("callback", callback)

Litestar

These methods are included in the complete app.py server below.

from wsrpc_asgi import WebSocketAsync
class RPC(WebSocketAsync):
    pass


async def echo(socket, **params):
    return params


async def callback(socket, *, value):
    return await socket.proxy.double(value=value)


RPC.add_route("echo", echo)
RPC.add_route("callback", callback)

Falcon

These methods are included in the complete app.py server below.

from wsrpc_asgi import WebSocketAsync
class RPC(WebSocketAsync):
    pass


async def echo(socket, **params):
    return params


async def callback(socket, *, value):
    return await socket.proxy.double(value=value)


RPC.add_route("echo", echo)
RPC.add_route("callback", callback)

Sanic

These methods are included in the complete app.py server below.

from wsrpc_asgi import WebSocketAsync
class RPC(WebSocketAsync):
    pass


async def echo(socket, **params):
    return params


async def callback(socket, *, value):
    return await socket.proxy.double(value=value)


RPC.add_route("echo", echo)
RPC.add_route("callback", callback)

BlackSheep

These methods are included in the complete app.py server below.

from wsrpc_asgi import WebSocketAsync
class RPC(WebSocketAsync):
    pass


async def echo(socket, **params):
    return params


async def callback(socket, *, value):
    return await socket.proxy.double(value=value)


RPC.add_route("echo", echo)
RPC.add_route("callback", callback)

Bare ASGI

These methods are included in the complete app.py server below.

from wsrpc_asgi import WebSocketAsync
class RPC(WebSocketAsync):
    pass


async def echo(socket, **params):
    return params


async def callback(socket, *, value):
    return await socket.proxy.double(value=value)


RPC.add_route("echo", echo)
RPC.add_route("callback", callback)

Connect your server

Save the selected server as app.py. Every tab is a complete, standalone file containing its RPC methods and framework integration. Start the server with the command in the same tab. All examples listen at ws://127.0.0.1:8000/ws/; all except bare ASGI also serve GET /health as an ordinary HTTP route.

aiohttp

from wsrpc_aiohttp import WebSocketAsync

from aiohttp import web


class RPC(WebSocketAsync):
    pass


async def echo(socket, **params):
    return params


async def callback(socket, *, value):
    return await socket.proxy.double(value=value)


RPC.add_route("echo", echo)
RPC.add_route("callback", callback)


async def health(request):
    return web.json_response({"ok": True})


def create_app():
    app = web.Application()
    app.router.add_get("/ws/", RPC)
    app.router.add_get("/health", health)
    return app


if __name__ == "__main__":
    web.run_app(create_app(), host="127.0.0.1", port=8000)

python app.py

FastAPI

serve() accepts an unaccepted FastAPI WebSocket. Dependencies run once per connection; their results are available through socket.context.

from fastapi import Depends, FastAPI, WebSocket
from wsrpc_asgi import WebSocketAsync


class RPC(WebSocketAsync):
    pass


async def echo(socket, **params):
    return params


async def callback(socket, *, value):
    return await socket.proxy.double(value=value)


RPC.add_route("echo", echo)
RPC.add_route("callback", callback)


app = FastAPI()


async def connection_context(websocket: WebSocket):
    return {"client": websocket.client.host if websocket.client else None}


@app.websocket("/ws/")
async def rpc(websocket: WebSocket, context=Depends(connection_context)):
    await RPC.serve(websocket, context=context)


@app.get("/health")
async def health():
    return {"ok": True}

uvicorn app:app --host 127.0.0.1 --port 8000

Starlette

Keep the endpoint and close it in the application’s lifespan, as shown here.

from contextlib import asynccontextmanager

from starlette.applications import Starlette
from starlette.responses import JSONResponse
from starlette.routing import Route, WebSocketRoute
from wsrpc_asgi import WebSocketAsync


class RPC(WebSocketAsync):
    pass


async def echo(socket, **params):
    return params


async def callback(socket, *, value):
    return await socket.proxy.double(value=value)


RPC.add_route("echo", echo)
RPC.add_route("callback", callback)


rpc = RPC.as_asgi()


@asynccontextmanager
async def lifespan(app):
    yield
    await rpc.close()


async def health(request):
    return JSONResponse({"ok": True})


app = Starlette(
    routes=[Route("/health", health), WebSocketRoute("/ws/", rpc)],
    lifespan=lifespan,
)

uvicorn app:app --host 127.0.0.1 --port 8000

Django

This standalone example configures minimal settings. In an existing project keep your DJANGO_SETTINGS_MODULE. The RPC wrapper bypasses Django HTTP middleware; authorize WebSockets separately.

import django
from django.conf import settings
from django.core.asgi import get_asgi_application
from django.http import JsonResponse
from django.urls import path
from wsrpc_asgi import RPCMiddleware, WebSocketAsync


class RPC(WebSocketAsync):
    pass


async def echo(socket, **params):
    return params


async def callback(socket, *, value):
    return await socket.proxy.double(value=value)


RPC.add_route("echo", echo)
RPC.add_route("callback", callback)


# In an existing project keep DJANGO_SETTINGS_MODULE and its settings.
if not settings.configured:
    settings.configure(
        DEBUG=True,
        SECRET_KEY="local-example-only",
        ROOT_URLCONF=__name__,
        ALLOWED_HOSTS=["*"],
    )
django.setup()


def health(request):
    return JsonResponse({"ok": True})


urlpatterns = [path("health", health)]
http_app = get_asgi_application()
rpc = RPC.as_asgi()
app = RPCMiddleware(http_app, rpc, path="/ws/")

uvicorn app:app --host 127.0.0.1 --port 8000

Channels

This standalone file includes Django settings, the HTTP application and the Channels router. In an existing Channels project retain its settings and authentication/origin middleware.

import django
from channels.routing import ProtocolTypeRouter, URLRouter
from django.conf import settings
from django.core.asgi import get_asgi_application
from django.http import JsonResponse
from django.urls import path
from wsrpc_asgi import WebSocketAsync


class RPC(WebSocketAsync):
    pass


async def echo(socket, **params):
    return params


async def callback(socket, *, value):
    return await socket.proxy.double(value=value)


RPC.add_route("echo", echo)
RPC.add_route("callback", callback)


# In an existing project keep DJANGO_SETTINGS_MODULE and its settings.
if not settings.configured:
    settings.configure(
        DEBUG=True,
        SECRET_KEY="local-example-only",
        ROOT_URLCONF=__name__,
        ALLOWED_HOSTS=["*"],
    )
django.setup()


def health(request):
    return JsonResponse({"ok": True})


urlpatterns = [path("health", health)]
http_app = get_asgi_application()
rpc = RPC.as_asgi()
app = ProtocolTypeRouter(
    {
        "http": http_app,
        "websocket": URLRouter([path("ws/", rpc)]),
    }
)

uvicorn app:app --host 127.0.0.1 --port 8000

Quart

from quart import Quart
from wsrpc_asgi import RPCMiddleware, WebSocketAsync


class RPC(WebSocketAsync):
    pass


async def echo(socket, **params):
    return params


async def callback(socket, *, value):
    return await socket.proxy.double(value=value)


RPC.add_route("echo", echo)
RPC.add_route("callback", callback)


app = Quart(__name__)


@app.get("/health")
async def health():
    return {"ok": True}


rpc = RPC.as_asgi()
app.asgi_app = RPCMiddleware(app.asgi_app, rpc, path="/ws/")

uvicorn app:app --host 127.0.0.1 --port 8000

Litestar

from litestar import Litestar, get
from wsrpc_asgi import RPCMiddleware, WebSocketAsync


class RPC(WebSocketAsync):
    pass


async def echo(socket, **params):
    return params


async def callback(socket, *, value):
    return await socket.proxy.double(value=value)


RPC.add_route("echo", echo)
RPC.add_route("callback", callback)


@get("/health")
async def health() -> dict[str, bool]:
    return {"ok": True}


http_app = Litestar(route_handlers=[health])
rpc = RPC.as_asgi()
app = RPCMiddleware(http_app, rpc, path="/ws/")

uvicorn app:app --host 127.0.0.1 --port 8000

Falcon

import falcon.asgi
from wsrpc_asgi import RPCMiddleware, WebSocketAsync


class RPC(WebSocketAsync):
    pass


async def echo(socket, **params):
    return params


async def callback(socket, *, value):
    return await socket.proxy.double(value=value)


RPC.add_route("echo", echo)
RPC.add_route("callback", callback)


class Health:
    async def on_get(self, request, response):
        response.media = {"ok": True}


http_app = falcon.asgi.App()
http_app.add_route("/health", Health())
rpc = RPC.as_asgi()
app = RPCMiddleware(http_app, rpc, path="/ws/")

uvicorn app:app --host 127.0.0.1 --port 8000

Sanic

Run through Uvicorn in ASGI mode so the RPC wrapper receives connections.

from sanic import Sanic
from sanic.response import json
from wsrpc_asgi import RPCMiddleware, WebSocketAsync


class RPC(WebSocketAsync):
    pass


async def echo(socket, **params):
    return params


async def callback(socket, *, value):
    return await socket.proxy.double(value=value)


RPC.add_route("echo", echo)
RPC.add_route("callback", callback)


http_app = Sanic("wsrpc_example")


@http_app.get("/health")
async def health(request):
    return json({"ok": True})


rpc = RPC.as_asgi()
app = RPCMiddleware(http_app, rpc, path="/ws/")

uvicorn app:app --host 127.0.0.1 --port 8000

BlackSheep

from blacksheep import Application
from blacksheep.server.responses import json
from wsrpc_asgi import RPCMiddleware, WebSocketAsync


class RPC(WebSocketAsync):
    pass


async def echo(socket, **params):
    return params


async def callback(socket, *, value):
    return await socket.proxy.double(value=value)


RPC.add_route("echo", echo)
RPC.add_route("callback", callback)


http_app = Application()


@http_app.router.get("/health")
async def health():
    return json({"ok": True})


rpc = RPC.as_asgi()
app = RPCMiddleware(http_app, rpc, path="/ws/")

uvicorn app:app --host 127.0.0.1 --port 8000

Bare ASGI

The endpoint itself is an ASGI callable. Use an asyncio server backend.

from wsrpc_asgi import WebSocketAsync


class RPC(WebSocketAsync):
    pass


async def echo(socket, **params):
    return params


async def callback(socket, *, value):
    return await socket.proxy.double(value=value)


RPC.add_route("echo", echo)
RPC.add_route("callback", callback)


rpc = RPC.as_asgi(path="/ws/")
app = rpc

uvicorn app:app --host 127.0.0.1 --port 8000

Each handler receives the current RPC connection as its first argument. Expose only the methods you intend peers to call. See ASGI integration details for authentication, lifecycle, server backends and deployment; aiohttp routes cover aiohttp-specific handler behavior.

Call from Python

Use either Python client with any server above. Save as client.py, then run python client.py while the server is running. The callback is registered before connecting, so the server can call it immediately.

aiohttp client

import asyncio
from wsrpc_aiohttp import WSRPCClient


async def main():
    async with WSRPCClient("ws://127.0.0.1:8000/ws/", timeout=5) as client:
        client.add_route("double", lambda socket, value: value * 2)
        print(await client.proxy.echo(message="Hello"))
        print(await client.proxy.callback(value=21))


if __name__ == "__main__":
    asyncio.run(main())

Expected output:

{'message': 'Hello'}
42

websockets client

import asyncio
from wsrpc_asgi import WSRPCClient


async def main():
    async with WSRPCClient("ws://127.0.0.1:8000/ws/", timeout=5) as client:
        client.add_route("double", lambda socket, value: value * 2)
        print(await client.proxy.echo(message="Hello"))
        print(await client.proxy.callback(value=21))


if __name__ == "__main__":
    asyncio.run(main())

Expected output:

{'message': 'Hello'}
42

Install wsrpc-aiohttp for the aiohttp client, or wsrpc-asgi[client] for the websockets client. Calls use keyword arguments. The context manager connects on entry and closes the connection on exit. Use wss:// with TLS.

Call from a browser

Both packages include the same JavaScript client. Serve STATIC_DIR at /js/ using your framework’s static-file support, then use the same code with every server:

<script src="/js/wsrpc.min.js"></script>
<script>
  const scheme = location.protocol === "https:" ? "wss:" : "ws:";
  const rpc = new WSRPC(`${scheme}//${location.host}/ws/`);
  rpc.addRoute("double", ({ value }) => value * 2);
  rpc.connect();

  async function run() {
    await rpc.onEvent("onconnect");
    try {
      console.log(await rpc.call("echo", { message: "Hello" }));
      console.log(await rpc.call("callback", { value: 21 }));
    } catch (error) {
      console.error(error);
    }
  }
  run();
</script>

For complete demos that serve both the page and JavaScript, see the aiohttp browser demo and FastAPI browser demo. Continue with clients, the protocol or API differences.