Warning

This is the documentation for version 5, released August 16, 2021. Read a post explaining the changes in version 5.

Welcome to jsonrpcserver’s documentation!¶

Dispatch JSON-RPC requests to your own functions, and get a response to send back.

Quickstart¶

Create a server.py:

from jsonrpcserver import Success, method, serve

@method
def ping():
    return Success("pong")

if __name__ == "__main__":
    serve()

Start the server:

$ pip install jsonrpcserver
$ python server.py
 * Listening on port 5000

Test the server:

$ curl -X POST http://localhost:5000 -d '{"jsonrpc": "2.0", "method": "ping", "id": 1}'
{"jsonrpc": "2.0", "result": "pong", "id": 1}

Methods¶

Methods are functions that can be called by a JSON-RPC request. To write one, decorate a function with @method:

from jsonrpcserver import method, Result, Success, Error

@method
def ping() -> Result:
    return Success("pong")

If you don’t need to respond with any value simply return Success().

Responses¶

Methods return either Success or Error. These are the JSON-RPC response objects (excluding the jsonrpc and id parts). Error takes a code, message, and optionally ‘data’.

@method
def test() -> Result:
    return Error(1, "There was a problem")

Note

Alternatively, raise a JsonRpcError, which takes the same arguments as Error.

Parameters¶

Methods can accept arguments.

@method
def hello(name: str) -> Result:
    return Success("Hello " + name)

Testing it:

$ curl -X POST http://localhost:5000 -d '{"jsonrpc": "2.0", "method": "hello", "params": ["Beau"], "id": 1}'
{"jsonrpc": "2.0", "result": "Hello Beau", "id": 1}

Invalid params¶

A common error response is invalid params. The JSON-RPC error code for this is -32602. A shortcut, InvalidParams, is included so you don’t need to remember that.

from jsonrpcserver import method, Result, InvalidParams, Success, dispatch

@method
def within_range(num: int) -> Result:
    if num not in range(1, 5):
        return InvalidParams("Value must be 1-5")
    return Success()

This is the same as saying

return Error(-32602, "Invalid params", "Value must be 1-5")

Dispatch¶

The dispatch function takes a JSON-RPC request, calls the appropriate method and gives a JSON-RPC response.

>>> dispatch('{"jsonrpc": "2.0", "method": "ping", "id": 1}')
'{"jsonrpc": "2.0", "result": "pong", "id": 1}'

See how dispatch is used in different frameworks.

Optional parameters¶

methods¶

This lets you specify a group of methods to dispatch to. It’s an alternative to using the @method decorator. The value should be a dict mapping function names to functions.

def ping():
    return Success("pong")

dispatch(request, methods={"ping": ping})

Default is global_methods, which is an internal dict populated by the @method decorator.

context¶

If specified, this will be the first argument to all methods.

@method
def greet(context, name):
    return Success(context + " " + name)

>>> dispatch('{"jsonrpc": "2.0", "method": "greet", "params": ["Beau"], "id": 1}', context="Hello")
'{"jsonrpc": "2.0", "result": "Hello Beau", "id": 1}'

deserializer¶

A function that parses the request string. Default is json.loads.

dispatch(request, deserializer=ujson.loads)

serializer¶

A function that serializes the response string. Default is json.dumps.

dispatch(request, serializer=ujson.dumps)

validator¶

A function that validates the request once the json has been parsed. The function should raise an exception (any exception) if the request doesn’t match the JSON-RPC spec. Default is default_validator which validates the request against a schema.

Async¶

Async dispatch is supported.

from jsonrpcserver import method, Success, async_dispatch

@method
async def ping() -> Result:
    return Success("pong")

await async_dispatch('{"jsonrpc": "2.0", "method": "ping", "id": 1}')

Some reasons to use this:

Use it with an asynchronous protocol like sockets or message queues.
await long-running functions from your method.
Batch requests are dispatched concurrently.

Notifications¶

Notifications are requests without an id. We should not respond to notifications, so jsonrpcserver gives an empty string to signify there is no response.

>>> await async_dispatch('{"jsonrpc": "2.0", "method": "ping"}')
''

If the response is an empty string, don’t send it.

if response := dispatch(request):
    send(response)

Note

A synchronous protocol like HTTP requires a response no matter what, so we can send back the empty string. However with async protocols, we have the choice of responding or not.

FAQ¶

How to disable schema validation?¶

Validating requests is costly - roughly 40% of dispatching time is spent on schema validation. If you know the incoming requests are valid, you can disable the validation for better performance.

dispatch(request, validator=lambda _: None)

Which HTTP status code to respond with?¶

I suggest:

200 if response else 204

If the request was a notification, dispatch will give you an empty string. So since there’s no http body, use status code 204 - no content.

How to rename a method¶

Use @method(name="new_name").

Or use dispatch function’s methods parameter.

How to get the response in other forms?¶

(Todo)

Examples ¶

Contents

Examples
- aiohttp
- Django
- FastAPI
- Flask
- http.server
- jsonrpcserver
- Sanic
- Socket.IO
- Tornado
- Websockets
- Werkzeug
- ZeroMQ
- ZeroMQ (asynchronous)

aiohttp ¶

from aiohttp import web
from jsonrpcserver import method, Result, Success, async_dispatch


@method
async def ping() -> Result:
    return Success("pong")


async def handle(request):
    return web.Response(
        text=await async_dispatch(await request.text()), content_type="application/json"
    )


app = web.Application()
app.router.add_post("/", handle)

if __name__ == "__main__":
    web.run_app(app, port=5000)

See blog post.

Django ¶

Create a views.py:

from django.http import HttpResponse
from django.views.decorators.csrf import csrf_exempt
from jsonrpcserver import method, Result, Success, dispatch


@method
def ping() -> Result:
    return Success("pong")


@csrf_exempt
def jsonrpc(request):
    return HttpResponse(
        dispatch(request.body.decode()), content_type="application/json"
    )

See blog post.

FastAPI ¶

from fastapi import FastAPI, Request, Response
from jsonrpcserver import Result, Success, dispatch, method
import uvicorn

app = FastAPI()


@method
def ping() -> Result:
    return Success("pong")


@app.post("/")
async def index(request: Request):
    return Response(dispatch(await request.body()))


if __name__ == "__main__":
    uvicorn.run(app, port=5000)

See blog post.

Flask ¶

from flask import Flask, Response, request
from jsonrpcserver import method, Result, Success, dispatch

app = Flask(__name__)


@method
def ping() -> Result:
    return Success("pong")


@app.route("/", methods=["POST"])
def index():
    print(request.get_data().decode())
    return Response(
        dispatch(request.get_data().decode()), content_type="application/json"
    )


if __name__ == "__main__":
    app.run()

See blog post.

http.server ¶

Using Python’s built-in http.server module.

from http.server import BaseHTTPRequestHandler, HTTPServer

from jsonrpcserver import method, Result, Success, dispatch


@method
def ping() -> Result:
    return Success("pong")


class TestHttpServer(BaseHTTPRequestHandler):
    def do_POST(self):
        # Process request
        request = self.rfile.read(int(self.headers["Content-Length"])).decode()
        response = dispatch(request)
        # Return response
        self.send_response(200)
        self.send_header("Content-type", "application/json")
        self.end_headers()
        self.wfile.write(response.encode())


if __name__ == "__main__":
    HTTPServer(("localhost", 5000), TestHttpServer).serve_forever()

See blog post.

jsonrpcserver ¶

Using jsonrpcserver’s built-in serve method.

from jsonrpcserver import method, Result, Success, serve


@method
def ping() -> Result:
    return Success("pong")


if __name__ == "__main__":
    serve()

Sanic ¶

from sanic import Sanic
from sanic.request import Request
from sanic.response import json
from jsonrpcserver import Result, Success, dispatch_to_serializable, method

app = Sanic("JSON-RPC app")


@method
def ping() -> Result:
    return Success("pong")


@app.route("/", methods=["POST"])
async def test(request: Request):
    return json(dispatch_to_serializable(request.body))


if __name__ == "__main__":
    app.run(port=5000)

See blog post.

Socket.IO ¶

from flask import Flask
from flask_socketio import SocketIO, send
from jsonrpcserver import method, Result, Success, dispatch

app = Flask(__name__)
socketio = SocketIO(app)


@method
def ping() -> Result:
    return Success("pong")


@socketio.on("message")
def handle_message(request):
    if response := dispatch(request):
        send(response, json=True)


if __name__ == "__main__":
    socketio.run(app, port=5000)

See blog post.

Tornado ¶

from jsonrpcserver import method, Result, Success, async_dispatch
from tornado import ioloop, web


@method
async def ping() -> Result:
    return Success("pong")


class MainHandler(web.RequestHandler):
    async def post(self) -> None:
        request = self.request.body.decode()
        if response := await async_dispatch(request):
            self.write(response)


app = web.Application([(r"/", MainHandler)])

if __name__ == "__main__":
    app.listen(5000)
    ioloop.IOLoop.current().start()

See blog post.

Websockets ¶

import asyncio

from jsonrpcserver import method, Success, Result, async_dispatch
import websockets


@method
async def ping() -> Result:
    return Success("pong")


async def main(websocket, path):
    if response := await async_dispatch(await websocket.recv()):
        await websocket.send(response)


start_server = websockets.serve(main, "localhost", 5000)
asyncio.get_event_loop().run_until_complete(start_server)
asyncio.get_event_loop().run_forever()

See blog post.

Werkzeug ¶

from jsonrpcserver import method, Result, Success, dispatch
from werkzeug.serving import run_simple
from werkzeug.wrappers import Request, Response


@method
def ping() -> Result:
    return Success("pong")


@Request.application
def application(request):
    return Response(dispatch(request.data.decode()), 200, mimetype="application/json")


if __name__ == "__main__":
    run_simple("localhost", 5000, application)

See blog post.

ZeroMQ ¶

from jsonrpcserver import method, Result, Success, dispatch
import zmq

socket = zmq.Context().socket(zmq.REP)


@method
def ping() -> Result:
    return Success("pong")


if __name__ == "__main__":
    socket.bind("tcp://*:5000")
    while True:
        request = socket.recv().decode()
        socket.send_string(dispatch(request))

See blog post.

ZeroMQ (asynchronous)¶

from jsonrpcserver import method, Result, Success, async_dispatch
import aiozmq
import asyncio
import zmq


@method
async def ping() -> Result:
    return Success("pong")


async def main():
    rep = await aiozmq.create_zmq_stream(zmq.REP, bind="tcp://*:5000")
    while True:
        request = (await rep.read())[0].decode()
        if response := (await async_dispatch(request)).encode():
            rep.write((response,))


if __name__ == "__main__":
    asyncio.set_event_loop_policy(aiozmq.ZmqEventLoopPolicy())
    asyncio.get_event_loop().run_until_complete(main())

See blog post.