serverwamp

Adds Web Application Messaging Protocol features to WebSocket servers. serverwamp makes it easy to both serve requests from clients and push events to clients over a single connection. Currently supports ASGI servers using asyncio or Trio and aiohttp servers.

Documentation

Usage Example

import asyncio

from aiohttp import web
import serverwamp

docs_api = serverwamp.RPCRouteSet()
docs_events = serverwamp.TopicRouteSet()

@docs_api.route('docs.getDoc')
async def get_doc(document_id):
    record = await my_db.retrieve(document_id)
    return {'status': 'SUCCESS', 'doc': record}

@docs_api.route('docs.deleteDoc')
async def delete_doc(document_id):
    succeeded = await my_db.delete(document_id)
    if succeeded:
        return {'status': 'SUCCESS'}

    return serverwamp.RPCErrorResult(
        'wamp.error.delete_failed',
        kwargs={'status': 'FAILURE'}
    )

@docs_events.route('docs.changes')
async def subscribe_to_doc_changes(session, changes_subscribers):
    changes_subscribers.add(session)
    session.send_event(args=('Thanks for subscribing!',))
    yield
    changes_subscribers.remove(session)


async def server_context(app):

    async def watch_for_docs_changes():
        async for event in my_db.watch_changes():
            for session in app['changes_subscribers']:
                await session.send_event(f'Document changed: {event}')

    watcher = asyncio.create_task(watch_for_docs_changes())
    try:
        yield
    finally:
        watcher.cancel()


if __name__ == '__main__':
    changes_subscribers = set()

    app = serverwamp.Application()
    app.add_rpc_routes(docs_api)
    app.add_topic_routes(docs_events)
    app.set_default_arg('changes_subscribers', changes_subscribers)

    http_app = web.Application()
    http_app['changes_subscribers'] = changes_subscribers
    http_app.add_routes((
        web.get('/', app.aiohttp_websocket_handler()),
    ))
    http_app.cleanup_ctx.append(server_context)

    web.run_app(http_app)

Development

Unit tests can be run with:

pip install -e . --group ci
pytest

Metadata

Release files for serverwamp 1.2.0

For a detailed explanation of source distributions (sdists) and built distributions (wheels), please see the package formats documentation.

Source distribution (sdist)

Source distribution for serverwamp 1.2.0
File	Size	Uploaded
serverwamp-1.2.0.tar.gz	23.0 kB	Sep 27, 2026	Details

Built distribution (wheel)

Table of built distributions (wheels) for serverwamp 1.2.0
File	Interpreter	ABI	Platform	Reset
serverwamp-1.2.0-py3-none-any.whl 27.7 kB Sep 27, 2026	Python 3	none	any	Details

Total release size: 50.7 kB

Release files / serverwamp-1.2.0.tar.gz

Download URL	serverwamp-1.2.0.tar.gz
Size	23.0 kB
Tags	Source
SHA-256 checksum How to use checksums	`ad9c702b300230abb7ac6a77ecb011166d7119f5ab05f7b2c0f6c2c8dcc8ba61`
BLAKE2b-256 checksum How to use checksums	`945acfaed8a65c9a3bb844dc8f802cd610fc44ffc3ca616a47add9c91d580751`
Upload date	Sep 27, 2026
Uploaded using Trusted Publishing? What is trusted publishing?	No
Uploaded via	`twine/7.0.0 CPython/3.14.3`

Release files / serverwamp-1.2.0-py3-none-any.whl

Download URL	serverwamp-1.2.0-py3-none-any.whl
Size	27.7 kB
Tags	Python 3
SHA-256 checksum How to use checksums	`8f7a08e3dbf87d7fb9a25b61c56b5cf5a2187915fe58e4aa77d0a82a58579e95`
BLAKE2b-256 checksum How to use checksums	`6f20ffd2fbe308a38270c6669c2854d0894920cccf93ea448eb8338d0de1645d`
Upload date	Sep 27, 2026
Uploaded using Trusted Publishing? What is trusted publishing?	No
Uploaded via	`twine/7.0.0 CPython/3.14.3`