Skip to content
Add JSON-RPC ability to channels
Branch: master
Clone or download
Type Name Latest commit message Commit time
Failed to load latest commit information.
example add test for no feedack from notification Apr 27, 2017
.gitignore Squashed commit of the following: Jan 27, 2017
.travis.yml Add code climate Apr 11, 2017
LICENSE Restructure folders a bit better Jan 27, 2017 Restructure folders a bit better Jan 27, 2017 Added readme for transport-specific parameters Apr 27, 2017
requirements.txt Added some extra classifiers and force channels v1 Apr 22, 2018 Added some extra classifiers and force channels v1 Apr 22, 2018
tox.ini Contant message processing via single __process() method (improves No… Apr 26, 2017


PyPI version Codacy Badge Build Status Coverage Status Code Climate

The Django-channels-jsonrpc is aimed to enable JSON-RPC functionnality on top of the excellent django channels project and especially their Websockets functionality. It is aimed to be:

  • Fully integrated with Channels
  • Fully implement JSON-RPC 1 and 2 protocol
  • Support both WebSocket and HTTP transports
  • Easy integration


The only Django-channels-jsonrpc dependency is the Django channels project


Download and extract the latest pre-built release.

Install the dependencies and devDependencies and start the server.

$ pip install django-channels-jsonrpc


See complete example here, and in particular

It is intended to be used as a Websocket consumer. See documentation except... simplier...

Import JsonRpcConsumer class and create the consumer

from channels_jsonrpc import JsonRpcConsumer

class MyJsonRpcConsumer(JsonRpcConsumer):

    def connect(self, message, **kwargs):
        Perform things on WebSocket connection start
        self.message.reply_channel.send({"accept": True})

        # Do stuff if needed

    def disconnect(self, message, **kwargs):
        Perform things on WebSocket connection close
        # Do stuff if needed

JsonRpcConsumer derives from Channels WebSocketConsumer, you can read about all it's features here:

Then the last step is to create the RPC methos hooks. IT is done with the decorator:


Like this:

def ping():
    return "pong"

MyJsonRpcConsumer.rpc_method() accept a string as a parameter to 'rename' the function

def ping():
    return "pong"

Will now be callable with "method":"" in the rpc call:

{"id":1, "jsonrpc":"2.0","method":"","params":{}}

RPC methods can obviously accept parameters. They also return "results" or "errors":

def ping(fake_an_error):
    if fake_an_error:
        # Will return an error to the client
        #  --> {"id":1, "jsonrpc":"2.0","method":"","params":{}}
        #  <-- {"id": 1, "jsonrpc": "2.0", "error": {"message": "fake_error", "code": -32000, "data": ["fake_error"]}}
        raise Exception("fake_error")
        # Will return a result to the client
        #  --> {"id":1, "jsonrpc":"2.0","method":"","params":{}}
        #  <-- {"id": 1, "jsonrpc": "2.0", "result": "pong"}
        return "pong"

Sessions and other parameters from Message object

The original channel message - that can contain sessions (if activated with http_user) and other important info can be easily accessed by retrieving the **kwargs and get a parameter named original_message

def json_rpc_method(param1, **kwargs):
    original_message = kwargs["orginal_message"]
    ##do something with original_message


class MyJsonRpcConsumerTest(JsonRpcConsumer):
    # Set to True to automatically port users from HTTP cookies
    # (you don't need channel_session_user, this implies it)
    http_user = True


    def ping(**kwargs):
        original_message = kwargs["orginal_message"]
        original_message.channel_session["test"] = True
        return "pong"


Inbound notifications

Those are the one sent from the client to the server. They are dealt with the same way RPC methods are, except that instead of using rpc_method(), you can use rpc_notification() Thos rpc_notifications can also retrieve the original_message object

# Will be triggered when receiving this
#  --> {"jsonrpc":"2.0","method":"notification.alt_name","params":["val_param1", "val_param2"]}
def notification1(param1, param2, **kwargs):
    original_message = kwargs["orginal_message"]
    # Do something with notification
    # ...
    # Notification shouldn't return anything.

Outbound notifications

The server might want to send notifications to one or more of its clients. For that JsonRpcWebsocketConsumer provides 2 static methods:

  • JsonRpcWebsocketConsumer.notify_group(group_name, method, params)

Using channels'groups you can notify a whole group using this method

def send_to_group(group_name):
    MyJsonRpcWebsocketConsumerTest.notify_group(group_name, "notification.notif", {"payload": 1234})
    return True

Calling the RPC-method will send this notification to all the group group_name

  • JsonRpcWebsocketConsumer.notify_channel(reply_channel, method, params)

This will notify only one channel/client.

def send_to_reply_channel(**kwargs):
    original_message = kwarg["original_message"]
                                                {"payload": 12})
    return True

The reply_channel can be found in theoriginal_message object.

Transport-specific rpc-method/notifications

If you want to restrict rpc methods or notifications access to a specific transport method (http or websocket) The two decorator rpc_method() and rpc_notification() accept parameters to restric their use. websocket (default: True) and http (default: True)

You can use them like this:

@MyJsonRpcWebsocketConsumerTest.rpc_notification("notification.alt_name", websocket=True, http=False)
def notification1(param1, param2, **kwargs):
    original_message = kwargs["orginal_message"]
    # This notification will only be used when using websocket transport

Custom JSON encoder class

from django.core.serializers.json import DjangoJSONEncoder

class DjangoJsonRpcConsumer(JsonRpcConsumer):
    json_encoder_class = DjangoJSONEncoder


The JsonRpcConsumer class can be tested the same way Channels Consumers are tested. See here

You just need to remember to set your JsonRpcConsumer class to TEST_MODE in the test:

from channels.tests import ChannelTestCase, HttpClient
from .consumer import MyJsonRpcConsumer

MyJsonRpcConsumer.TEST_MODE = True

class TestsJsonConsumer(ChannelTestCase):
    def assertResult(self, method, params, result, error=False):
        client = HttpClient()
        client.send_and_consume('websocket.receive', text=request(method, params))
        key = "result" if not error else "error"
        message = client.receive()
        if message is None or key not in message:
            raise KeyError("'%s' key not in message: %s" % (key, message))

        self.assertEquals(message[key], result)

    def assertError(self, method, params, result):
        self.assertResult(method, params, result, True)

    def test_assert_result(self):

         self.assertResult("ping", {}, "pong")



Have fun with Websockets!

Free Software, Hell Yeah!

You can’t perform that action at this time.