openrpc 1.2.10

OpenRPC provides classes to rapidly develop an OpenRPC server.

OpenRPC

OpenRPC provides classes to rapidly develop an OpenRPC server.

Installation

OpenRPC is on PyPI and can be installed with:

pip install openrpc

Usage

This library provides an OpenRPCServer class that can be used to quickly create an OpenRPCServer; it takes as an argument an InfoObject which needs at minimum a title and version.

from openrpc.objects import InfoObject
from openrpc.server import OpenRPCServer

rpc = OpenRPCServer(InfoObject(title="Demo Server", version="1.0.0"))

Register a function as an RPC Method

To register a method with the OpenRPCServer add the @rpc.method decorator to a method.

@rpc.method
def add(a: int, b: int) -> int:
    return a + b

RPC Discover

The rpc.discover method is automatically generated. It relies heavily on type hints.

Process JSON RPC Request

OpenRPC is transport agnostic. To use it, pass JSON RPC requests to the process_request method.

req = """
{
  "id": 1,
  "method": "add",
  "params": {"a": 2, "b": 2},
  "jsonrpc": "2.0"
}
"""
rpc.process_request(req)  # '{"id": 1, "result": 4, "jsonrpc": "2.0}'

Async Support (v1.2+)

OpenRPC has async support:

await rpc.process_request_async(req)

Pydantic Support

For data classes, it is strongly recommended you use Pydantic. OpenRPCServer will use Pydantic for JSON serialization/deserialization as well as generating schemas.

Example

from flask import Flask, Response, request
from openrpc.objects import InfoObject
from openrpc.server import OpenRPCServer
from pydantic import BaseModel

app = Flask(__name__)
rpc = OpenRPCServer(InfoObject(title="Demo Server", version="1.0.0"))


class Vector3(BaseModel):
    x: float
    y: float
    z: float


@rpc.method
def get_distance(a: Vector3, b: Vector3) -> Vector3:
    return Vector3(
        x=a.x - b.x,
        y=a.y - b.y,
        z=a.z - b.z,
    )


@rpc.method
def add(a: int, b: int) -> int:
    return a + b


@app.route("/api/v1/", methods=["POST"])
def process_rpc() -> str:
    return rpc.process_request(request.data)


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

Example In

[
  {
    "id": 1,
    "method": "add",
    "params": {"a": 1, "b": 3},
    "jsonrpc": "2.0"
  }, {
    "id": 2,
    "method": "add",
    "params": [5, 7],
    "jsonrpc": "2.0"
  }, {
    "id": 3,
    "method": "add",
    "params": [11, "thirteen"],
    "jsonrpc": "2.0"
  }
]

Example Result Out

[
  {
    "id": 1,
    "result": 4,
    "jsonrpc": "2.0"
  }, {
    "id": 2,
    "result": 12,
    "jsonrpc": "2.0"
  }, {
    "id": 3,
    "error": {
      "code": -32000,
      "message": "TypeError: unsupported operand type(s) for +: 'int' and 'str'"
    },
    "jsonrpc": "2.0"
  }
]

Example RPC Discover

{
  "id": 1,
  "result": {
    "openrpc": "1.2.6",
    "info": {
      "title": "Demo Server",
      "version": "1.0.0"
    },
    "methods": [
      {
        "name": "add",
        "params": [
          {
            "name": "a",
            "schema": {
              "type": "number"
            },
            "required": true
          },
          {
            "name": "b",
            "schema": {
              "type": "number"
            },
            "required": true
          }
        ],
        "result": {
          "name": "result",
          "schema": {
            "type": "number"
          },
          "required": true
        }
      }
    ],
    "components": {
      "schemas": {}
    }
  },
  "jsonrpc": "2.0"
}