API Reference
Installation
Main Classes
JSONRPC
Constructor:
JSONRPC(
version: str = '2.0',
validate_results: bool = False,
context_type: type | None = None,
allow_batch: bool | None = None,
allow_dict_params: bool | None = None,
allow_list_params: bool | None = None,
max_batch: int = 100,
max_concurrent: int | None = None,
)
| Parameter | Type | Default | Description |
|---|---|---|---|
version |
'1.0' | '2.0' |
'2.0' |
JSON-RPC protocol version |
validate_results |
bool |
False |
Validate return types against annotations |
context_type |
type \| None |
None |
Expected context type for validation |
allow_batch |
bool \| None |
None |
Batch requests (default: spec-compliant) |
allow_dict_params |
bool \| None |
None |
Object params (default: spec-compliant) |
allow_list_params |
bool \| None |
None |
Array params (default: spec-compliant) |
max_batch |
int |
100 |
Max requests per batch (-1 = unlimited) |
max_concurrent |
int \| None |
None |
Max concurrent coroutines in async batch (None = os.cpu_count(), -1 = unlimited) |
Default params mode (spec-compliant):
| Version | allow_batch |
allow_dict_params |
allow_list_params |
|---|---|---|---|
| v1.0 | False |
False |
True |
| v2.0 | True |
True |
True |
Methods:
# Handle JSON-RPC request (sync)
rpc.handle(raw_data: str | bytes, context: Any = None) -> str | None
# Handle JSON-RPC request (async)
await rpc.handle_async(raw_data: str | bytes, context: Any = None) -> str | None
# Direct method call (sync, no JSON)
rpc.call_method(
method: str,
params: dict | list | None = None,
context: Any = None,
validate_result: bool = False,
) -> Any
# Direct method call (async, no JSON)
await rpc.call_method_async(
method: str,
params: dict | list | None = None,
context: Any = None,
validate_result: bool = False,
) -> Any
# Register method or group
rpc.register(name: str | None, target: Method | MethodGroup) -> None
# Unregister method or group by name or dotted path
rpc.unregister(path: str) -> None
# Decorator for rapid prototyping (v2.0 only)
@rpc.method
@rpc.method("custom_name")
Overridable hooks:
# Deserialize incoming JSON bytes/string to a Python object.
# Called once per request, before routing or validation.
def deserialize(self, data: str | bytes) -> Any: ...
# Serialize a response dict (or list of dicts for batch) to a JSON string.
def serialize(self, data: Any) -> str: ...
# Convert the method's return value to a JSON-serializable object.
# Override to handle custom types (datetime, Decimal, UUID, etc.)
# before they reach the JSON serializer.
def serialize_result(self, result: Any) -> Any: ...
All three hooks work together. Overriding deserialize + serialize is sufficient for a
full JSON library swap:
import orjson
class FastRPC(JSONRPC):
def deserialize(self, data):
return orjson.loads(data)
def serialize(self, data):
return orjson.dumps(data).decode()
Override serialize_result to handle custom types before serialization:
from datetime import datetime
from decimal import Decimal
class MyRPC(JSONRPC):
def serialize_result(self, result):
if isinstance(result, datetime):
return result.isoformat()
if isinstance(result, Decimal):
return str(result)
return super().serialize_result(result)
parse_request / parse_response and json.loads
deserialize() is the only place json.loads is called in the server path.
The module-level parse_request() and parse_response() utilities also accept
raw JSON strings and call json.loads internally — but JSONRPC always passes
them an already-parsed dict, so they never call json.loads during request
handling. If you override deserialize(), the full server path is covered.
Examples:
rpc = JSONRPC(version='2.0')
response = rpc.handle('{"jsonrpc": "2.0", "method": "add", "params": {"a": 1, "b": 2}, "id": 1}')
# Async
response = await rpc.handle_async(request_bytes, context=ctx)
# Direct call
result = rpc.call_method('math.add', {'a': 1, 'b': 2})
Method
Base class for all RPC methods.
class MyMethod(Method):
def execute(self, params: ParamsType) -> ReturnType:
...
# Optional: async
async def execute(self, params: ParamsType) -> ReturnType:
...
# Optional: with context
def execute(self, params: ParamsType, context: ContextType) -> ReturnType:
...
Signature rules:
- params must be 2nd parameter (after self)
- params type hint must be a @dataclass or None
- Return type annotation is required
- Optional 3rd parameter context with type hint
Available in execute():
- self.rpc - access to parent JSONRPC instance for internal calls
Examples:
from dataclasses import dataclass
from jsonrpc import Method
@dataclass
class AddParams:
a: int
b: int
class Add(Method):
def execute(self, params: AddParams) -> int:
# Call another method internally
return self.rpc.call_method('other.method', {'x': params.a})
MethodGroup
Container for organizing methods with a common prefix.
group = MethodGroup()
# Register method
group.register('method_name', MethodInstance())
# Register subgroup
group.register('subgroup_name', SubGroup())
# Unregister method or subgroup (unified)
group.unregister('method_name')
group.unregister('subgroup_name')
Override for middleware:
class CustomGroup(MethodGroup):
def execute_method(self, method, params, context=None):
# Before execution
result = super().execute_method(method, params, context)
# After execution
return result
OpenAPIGenerator
Generates OpenAPI 3.0 schema from registered methods.
generator = OpenAPIGenerator(
rpc: JSONRPC,
base_url: str = "/jsonrpc",
title: str = "JSON-RPC API",
version: str = "1.0.0",
description: str | None = None,
simplify_id: bool = True,
)
| Parameter | Type | Default | Description |
|---|---|---|---|
rpc |
JSONRPC |
— | RPC instance with registered methods |
base_url |
str |
"/jsonrpc" |
Base URL path for the JSON-RPC endpoint |
title |
str |
"JSON-RPC API" |
API title in schema |
version |
str |
"1.0.0" |
API version in schema |
description |
str \| None |
None |
API description |
simplify_id |
bool |
True |
Use {"type": "integer"} for id fields. Set to False for spec-compliant oneOf with string and integer. |
Methods:
# Generate OpenAPI spec as dict
spec: dict = generator.generate()
# Generate as JSON string
json_str: str = generator.generate_json(indent=2)
# Generate as YAML string (requires PyYAML)
yaml_str: str = generator.generate_yaml()
# Add security scheme
generator.add_security_scheme(
name: str,
scheme_type: Literal["apiKey", "http", "oauth2", "openIdConnect"],
**kwargs,
)
# Add global header parameter
generator.add_header(
name: str,
description: str,
required: bool = False,
schema: dict | None = None,
)
# Add global security requirement
generator.add_security_requirement(
scheme_name: str,
scopes: list[str] | None = None,
)
Examples:
generator = OpenAPIGenerator(
rpc,
base_url="/rpc",
title="My API",
version="2.0.0",
)
generator.add_security_scheme(
"BearerAuth",
scheme_type="http",
scheme="bearer",
bearerFormat="JWT",
)
generator.add_security_requirement("BearerAuth")
spec = generator.generate()
Error Classes
from jsonrpc.errors import (
RPCError,
ParseError,
InvalidRequestError,
MethodNotFoundError,
InvalidParamsError,
InternalError,
InvalidResultError,
ServerError,
)
| Class | Code | When to use |
|---|---|---|
ParseError |
-32700 |
Invalid JSON |
InvalidRequestError |
-32600 |
Invalid request structure |
MethodNotFoundError |
-32601 |
Method does not exist |
InvalidParamsError |
-32602 |
Parameter validation failed |
InternalError |
-32603 |
Internal server error |
InvalidResultError |
-32001 |
Return type mismatch |
ServerError(msg, code) |
-32000 to -32099 |
Custom server errors |
Usage:
from jsonrpc.errors import InvalidParamsError, ServerError
class MyMethod(Method):
def execute(self, params: Params) -> dict:
if not params.email:
raise InvalidParamsError("Email is required")
if database_down():
raise ServerError("Database unavailable", code=-32010)
return {"ok": True}
Type System
Supported Parameter Types
| Python Type | JSON type | Notes |
|---|---|---|
int |
number (integer) | No float coercion |
float |
number | Accepts int values too |
str |
string | |
bool |
boolean | Must be exactly true/false |
list |
array | Untyped |
list[T] |
array | Each element validated as T |
dict |
object | Untyped |
dict[K, V] |
object | Keys and values validated |
@dataclass |
object | Recursively validated |
T \| None |
T or null | Optional type |
T \| U |
T or U | Union type (first match wins) |
Literal["a", "b"] |
enum string | Restricted values |
Any |
any | No validation |
None |
null or absent | No parameters |
Nested Type Example
@dataclass
class Config:
features: dict[str, bool]
mode: Literal["dev", "prod"]
tags: list[str] | None = None
max_workers: int = 4
Protocol Compliance
Strict Mode (Default)
# v1.0: arrays, no batch
rpc = JSONRPC(version='1.0')
# Accepts: {"method": "add", "params": [1, 2], "id": 1}
# v2.0: objects, batch allowed
rpc = JSONRPC(version='2.0')
# Accepts: {"jsonrpc": "2.0", "method": "add", "params": {"a": 1, "b": 2}, "id": 1}
Permissive Mode
rpc = JSONRPC(
version='2.0',
allow_batch=True, # Already default for v2.0
allow_dict_params=True, # Already default for v2.0
allow_list_params=True, # Already default for v2.0
)
Decorator API (v2.0 only)
Prototyping Only
The decorator API is designed for rapid prototyping. Use Method classes for production.
rpc = JSONRPC(version='2.0') # v2.0 required
# Simple decorator
@rpc.method
def add(a: int, b: int) -> int:
return a + b
# Custom method name
@rpc.method("my_custom_name")
def some_function(x: str) -> str:
return x.upper()
# Async
@rpc.method
async def fetch(url: str) -> dict:
return await http_get(url)
Rules:
- All parameters must have type hints
- Return type annotation required
- No context parameter support
- Only root-level registration
- Original function still callable: add(1, 2) returns 3
Complete Usage Example
from dataclasses import dataclass
from jsonrpc import JSONRPC, Method, MethodGroup
from jsonrpc.errors import InvalidParamsError
from jsonrpc.openapi import OpenAPIGenerator
# Parameters
@dataclass
class SearchParams:
query: str
limit: int = 10
@dataclass
class AuthContext:
user_id: int | None
# Methods
class Search(Method):
def execute(self, params: SearchParams, context: AuthContext) -> list[dict]:
"""Search items."""
if not context.user_id:
raise InvalidParamsError("Authentication required")
return [{"id": 1, "title": f"Result: {params.query}"}]
# Setup
rpc = JSONRPC(version='2.0', context_type=AuthContext)
group = MethodGroup()
group.register('items', Search())
rpc.register('search', group)
# OpenAPI
generator = OpenAPIGenerator(rpc, title="My API", version="1.0.0")
spec = generator.generate()
# Handle request
ctx = AuthContext(user_id=42)
response = rpc.handle(
'{"jsonrpc": "2.0", "method": "search.items", "params": {"query": "test"}, "id": 1}',
context=ctx
)
Changelog
0.3.1 (First Public Release)
- JSON-RPC 1.0 and 2.0 support
- Dataclass-based parameter validation
- Built-in OpenAPI generation
- Hierarchical context support
- Decorator API for prototyping
- Async/sync methods
- Batch request handling
- Strict mode by default
- Zero external dependencies