Warning

You’re reading the docs for the upcoming version 4. Use the sidebar to switch to another version.

Welcome to jsonrpcserver’s documentation!

Generate JSON-RPC requests and parse responses.

pip install "jsonrpcclient<5"

Generating requests

The request function

Generate a request with the request function:

>>> from jsonrpcclient import request
>>> request("ping")
{"jsonrpc": "2.0", "method": "ping", "2.0", "id": 1}

Ids

Subsequent calls increment the id:

>>> request("ping")
{"jsonrpc": "2.0", "method": "ping", "2.0", "id": 2}
>>> request("ping")
{"jsonrpc": "2.0", "method": "ping", "2.0", "id": 3}

Use an explicit id:

>>> request("ping", id="foo")
{"jsonrpc": "2.0", "method": "ping", "2.0", "id": "foo"}

Or generate a different type of id:

>>> from jsonrpcclient import request_hex, request_random, request_uuid
>>> request_hex("foo")
{"jsonrpc": "2.0", "method": "foo", "id": "1"}
>>> request_random("foo")
{"jsonrpc": "2.0", "method": "foo", "id": "qzsib147"}
>>> request_uuid("foo")
{"jsonrpc": "2.0", "method": "foo", "id": "45480a2f-069c-42aa-a67f-f6fdd83d6026"}

Parameters

Pass params to include parameters in the payload. This should be either a tuple for positional arguments, or dict for keyword arguments.

>>> request("ping", params=(1,))
{"jsonrpc": "2.0", "method": "ping", "2.0", "params": [1], "id": 4}
>>> request("ping", params={"key": "val"})
{"jsonrpc": "2.0", "method": "ping", "2.0", "params": {"key": "val"}, "id": 5}

JSON requests

If you need the request serialized to a string, use request_json:

>>> from jsonrpcclient import request_json
>>> request_json("foo")
'{"jsonrpc": "2.0", "method": "foo", "id": 6}'

You can also use request_json_hex etc., for the other id types.

Batch requests

>>> import json
>>> json.dumps([request("foo") for _ in range(3)])
'[{"jsonrpc": "2.0", "method": "foo", "id": 7}, {"jsonrpc": "2.0", "method": "foo", "id": 8}, {"jsonrpc": "2.0", "method": "foo", "id": 9}]'

Notifications

Use the notification function instead of request:

>>> from jsonrpcclient import notification
>>> notification("ping")
{"jsonrpc": "2.0", "method": "ping"}

Similar to request_json, notification_json will give you the notification as a JSON string.

>>> from jsonrpcclient import notification_json
>>> notification_json("ping")
'{"jsonrpc": "2.0", "method": "ping"}'

Parsing responses

The library includes a parse function which turns a deserialized response into a nice namedtuple.

>>> parse({"jsonrpc": "2.0", "result": "pong", "id": 1})
Ok(result='pong', id=1)
>>> parse({"jsonrpc": "2.0", "error": {"code": 1, "message": "There was an error", "data": None}, "id": 1})
Error(code=1, message='There was an error', data=None, id=1)

If you have a string, use parse_json.

>>> parse_json('{"jsonrpc": "2.0", "result": "pong", "id": 1}')
Ok(result='pong', id=1)

To use the result, in Python versions prior to 3.10:

from jsonrpcclient import Error, Ok
parsed = parse(response)
if isinstance(parsed, Ok):
    print(parsed.result)
else:
    logging.error(parsed.message)

In Python 3.10+, use pattern matching:

match parse(response):
    case Ok(result, id):
        print(result)
    case Error(code, message, data, id):
        logging.error(message)

Examples

Contents

• Examples

  • aiohttp

  • Requests

  • Websockets

  • ZeroMQ

aiohttp

import asyncio
import logging

from aiohttp import ClientSession
from jsonrpcclient import Ok, request, parse


async def main():
    async with ClientSession() as session:
        async with session.post(
            "http://localhost:5000", json=request("ping")
        ) as response:
            parsed = parse(await response.json())
            if isinstance(parsed, Ok):
                print(parsed.result)
            else:
                logging.error(parsed.message)


asyncio.get_event_loop().run_until_complete(main())

See blog post.

Requests

from jsonrpcclient import request, parse, Ok
import logging
import requests

response = requests.post("http://localhost:5000/", json=request("ping"))

parsed = parse(response.json())
if isinstance(parsed, Ok):
    print(parsed.result)
else:
    logging.error(parsed.message)

Websockets

import asyncio
import logging

from jsonrpcclient import Ok, parse_json, request_json
import websockets


async def main():
    async with websockets.connect("ws://localhost:5000") as ws:
        await ws.send(request_json("ping"))
        response = parse_json(await ws.recv())

    if isinstance(response, Ok):
        print(response.result)
    else:
        logging.error(response.message)


asyncio.get_event_loop().run_until_complete(main())

See blog post.

ZeroMQ

from jsonrpcclient import request_json, parse_json, Ok
import logging
import zmq

socket = zmq.Context().socket(zmq.REQ)
socket.connect("tcp://localhost:5000")
socket.send_string(request_json("ping"))

response = parse_json(socket.recv().decode())
if isinstance(response, Ok):
    print(response.result)
else:
    logging.error(response.message)

See blog post.