KITAPINI
KITAPINI Docs

JSON-RPC 2.0 with Python

Learn how to build JSON-RPC 2.0 servers and clients using Python's standard library and third-party packages.

Getting Started with JSON-RPC 2.0 in Python

JSON-RPC (JSON Remote Procedure Call) is a simple, lightweight remote procedure call protocol encoded in JSON. JSON-RPC 2.0 provides a standardized format for sending requests, notifications, and responses between a client and a server.

JSON-RPC 2.0 is transport-agnostic. While this guide demonstrates HTTP, the same JSON payloads can be sent over WebSockets, TCP sockets, or standard input/output (stdio).

1. What is JSON-RPC 2.0?

A JSON-RPC request represents a call to a remote function. A request object contains:

  • jsonrpc: A String specifying the version of the JSON-RPC protocol (MUST be "2.0").
  • method: A String containing the name of the method to be invoked.
  • params: A Structured value (Array or Object) holding parameter values.
  • id: An identifier established by the client (String, Number, or null). Omit this field for notifications.

Payload Specification Examples

{
  "jsonrpc": "2.0",
  "method": "subtract",
  "params": [42, 23],
  "id": 1
}
{
  "jsonrpc": "2.0",
  "result": 19,
  "id": 1
}
{
  "jsonrpc": "2.0",
  "error": {
    "code": -32601,
    "message": "Method not found"
  },
  "id": 1
}

2. Standard Library Implementation

You can build a lightweight JSON-RPC 2.0 server and client using only Python's standard library (http.server and urllib).

import json
from http.server import HTTPServer, BaseHTTPRequestHandler

METHODS = {}

def export_rpc(func):
    """Decorator to register a function as an RPC method."""
    METHODS[func.__name__] = func
    return func

@export_rpc
def add(a, b):
    return a + b

@export_rpc
def greet(name="World"):
    return f"Hello, {name}!"

class JSONRPCHandler(BaseHTTPRequestHandler):
    def do_POST(self):
        content_length = int(self.headers.get('Content-Length', 0))
        body = self.rfile.read(content_length).decode('utf-8')

        try:
            request = json.loads(body)
        except json.JSONDecodeError:
            self._send_response(self._make_error(-32700, "Parse error", None))
            return

        if request.get("jsonrpc") != "2.0" or "method" not in request:
            self._send_response(self._make_error(-32600, "Invalid Request", request.get("id")))
            return

        method_name = request.get("method")
        params = request.get("params", [])
        req_id = request.get("id")

        if method_name not in METHODS:
            self._send_response(self._make_error(-32601, "Method not found", req_id))
            return

        try:
            if isinstance(params, list):
                result = METHODS[method_name](*params)
            elif isinstance(params, dict):
                result = METHODS[method_name](**params)
            else:
                self._send_response(self._make_error(-32602, "Invalid params", req_id))
                return

            if req_id is not None:
                response = {
                    "jsonrpc": "2.0",
                    "result": result,
                    "id": req_id
                }
                self._send_response(response)
        except Exception as e:
            self._send_response(self._make_error(-32603, f"Internal error: {str(e)}", req_id))

    def _make_error(self, code, message, req_id):
        return {
            "jsonrpc": "2.0",
            "error": {"code": code, "message": message},
            "id": req_id
        }

    def _send_response(self, response_dict):
        self.send_response(200)
        self.send_header("Content-Type", "application/json-rpc")
        self.end_headers()
        self.wfile.write(json.dumps(response_dict).encode('utf-8'))

def run_server(port=8000):
    server = HTTPServer(('localhost', port), JSONRPCHandler)
    print(f"JSON-RPC Server running on port {port}...")
    server.serve_forever()

if __name__ == "__main__":
    run_server()

3. Using Third-Party Libraries

For production applications, standard packages like jsonrpcserver are recommended.

pip install jsonrpcserver requests
from http.server import HTTPServer, BaseHTTPRequestHandler
from jsonrpcserver import method, Result, Success, dispatch

@method
def multiply(a: int, b: int) -> Result:
    return Success(a * b)

class RequestHandler(BaseHTTPRequestHandler):
    def do_POST(self):
        request_data = self.rfile.read(int(self.headers["Content-Length"])).decode()
        response = dispatch(request_data)
        self.send_response(200)
        self.send_header("Content-Type", "application/json")
        self.end_headers()
        self.wfile.write(response.encode())

if __name__ == "__main__":
    server = HTTPServer(("localhost", 5000), RequestHandler)
    print("Server listening on http://localhost:5000...")
    server.serve_forever()

4. Key JSON-RPC 2.0 Error Codes

CodeMessageMeaning
-32700Parse errorInvalid JSON was received by the server.
-32600Invalid RequestThe JSON sent is not a valid Request object.
-32601Method not foundThe method does not exist or is not available.
-32602Invalid paramsInvalid method parameter(s).
-32603Internal errorInternal JSON-RPC error.
-32000 to -32099Server errorReserved for implementation-defined server errors.