tessl/pypi-json-rpc

JSON-RPC transport implementation for Python supporting both 1.0 and 2.0 protocols with Django and Flask backends

—

Pending

Overview

Eval results

Files

Requests and Responses

Name: tessl/pypi-json-rpc
Author: tessl

Structured request and response objects for JSON-RPC 1.0 and 2.0 protocols with automatic serialization, validation, and comprehensive support for both single and batch operations.

Capabilities

JSON-RPC 2.0 Requests

Modern JSON-RPC 2.0 request objects with support for named and positional parameters, notifications, and batch processing.

class JSONRPC20Request:
    JSONRPC_VERSION = "2.0"
    REQUIRED_FIELDS = {"jsonrpc", "method"}
    POSSIBLE_FIELDS = {"jsonrpc", "method", "params", "id"}
    
    def __init__(self, method: str = None, params = None, _id = None, is_notification: bool = None):
        """
        Create JSON-RPC 2.0 request.
        
        Parameters:
        - method: Method name to invoke (required)
        - params: Method parameters (list, dict, or None)
        - _id: Request identifier (string, number, or None)
        - is_notification: Whether this is a notification (no response expected)
        """
    
    @classmethod
    def from_json(cls, json_str: str):
        """Parse JSON string into request object."""
    
    @classmethod
    def from_data(cls, data):
        """Create request from parsed dict or list (for batch)."""
    
    # Properties
    method: str
    params  # list, dict, or None
    _id  # str, int, or None
    is_notification: bool
    data: dict
    json: str
    args: tuple  # Positional parameters as tuple
    kwargs: dict  # Named parameters as dict

JSON-RPC 1.0 Requests

Legacy JSON-RPC 1.0 request objects with array-only parameters and simplified structure.

class JSONRPC10Request:
    JSONRPC_VERSION = "1.0"
    REQUIRED_FIELDS = {"method", "params", "id"}
    POSSIBLE_FIELDS = {"method", "params", "id"}
    
    def __init__(self, method: str = None, params = None, _id = None, is_notification: bool = None):
        """
        Create JSON-RPC 1.0 request.
        
        Parameters:
        - method: Method name to invoke (required)
        - params: Method parameters (list or tuple only)
        - _id: Request identifier (any type except None for non-notifications)
        - is_notification: Whether this is a notification
        """
    
    @classmethod
    def from_json(cls, json_str: str):
        """Parse JSON string into request object."""
    
    @classmethod
    def from_data(cls, data: dict):
        """Create request from parsed dict."""
    
    # Properties
    method: str
    params: list
    _id
    is_notification: bool
    data: dict
    json: str
    args: tuple
    kwargs: dict

Batch Requests

JSON-RPC 2.0 batch request container for processing multiple requests in a single call.

class JSONRPC20BatchRequest:
    JSONRPC_VERSION = "2.0"
    
    def __init__(self, *requests):
        """
        Create batch request container.
        
        Parameters:
        - requests: Variable number of JSONRPC20Request objects
        """
    
    @classmethod
    def from_json(cls, json_str: str):
        """Parse JSON array string into batch request."""
    
    # Properties
    requests: tuple
    json: str
    
    def __iter__(self):
        """Iterate over individual requests."""

JSON-RPC 2.0 Responses

Response objects for JSON-RPC 2.0 with result or error data and proper correlation with requests.

class JSONRPC20Response:
    JSONRPC_VERSION = "2.0"
    
    def __init__(self, **kwargs):
        """
        Create JSON-RPC 2.0 response.
        
        Parameters:
        - result: Success result (mutually exclusive with error)  
        - error: Error object dict (mutually exclusive with result)
        - _id: Request ID for correlation
        """
    
    # Properties
    result  # Any type
    error: dict  # Error object
    _id  # str, int, or None
    data: dict
    json: str
    request  # Associated request object

JSON-RPC 1.0 Responses

Response objects for JSON-RPC 1.0 with simplified structure.

class JSONRPC10Response:
    JSONRPC_VERSION = "1.0"
    
    def __init__(self, **kwargs):
        """
        Create JSON-RPC 1.0 response.
        
        Parameters:
        - result: Success result (mutually exclusive with error)
        - error: Error object dict (mutually exclusive with result)  
        - _id: Request ID (required, cannot be None)
        """
    
    # Properties  
    result  # Any type
    error: dict
    _id  # Required, not None
    data: dict
    json: str
    request  # Associated request object

Batch Responses

Container for multiple JSON-RPC 2.0 responses from batch request processing.

class JSONRPC20BatchResponse:
    JSONRPC_VERSION = "2.0"
    
    def __init__(self, *responses):
        """
        Create batch response container.
        
        Parameters:
        - responses: Variable number of response objects
        """
    
    # Properties
    responses: tuple
    data: list
    json: str
    request  # Associated batch request
    
    def __iter__(self):
        """Iterate over individual responses."""

Usage Examples

Creating and Parsing Requests

from jsonrpc.jsonrpc2 import JSONRPC20Request
from jsonrpc.jsonrpc1 import JSONRPC10Request

# JSON-RPC 2.0 with named parameters
request = JSONRPC20Request(
    method="subtract", 
    params={"minuend": 42, "subtrahend": 23}, 
    _id=1
)
print(request.json)
# {"jsonrpc": "2.0", "method": "subtract", "params": {"minuend": 42, "subtrahend": 23}, "id": 1}

# JSON-RPC 2.0 with positional parameters  
request = JSONRPC20Request(
    method="subtract",
    params=[42, 23],
    _id=2
)
print(request.json)
# {"jsonrpc": "2.0", "method": "subtract", "params": [42, 23], "id": 2}

# JSON-RPC 2.0 notification (no response expected)
notification = JSONRPC20Request(
    method="update",
    params=[1, 2, 3, 4, 5],
    is_notification=True
)
print(notification.json)
# {"jsonrpc": "2.0", "method": "update", "params": [1, 2, 3, 4, 5]}

# JSON-RPC 1.0 request
request_v1 = JSONRPC10Request(
    method="subtract",
    params=[42, 23],
    _id=1
)
print(request_v1.json)
# {"method": "subtract", "params": [42, 23], "id": 1}

Parsing JSON Strings

from jsonrpc.jsonrpc import JSONRPCRequest

# Auto-detect version and parse
json_str = '{"jsonrpc": "2.0", "method": "add", "params": [1, 2], "id": 1}'
request = JSONRPCRequest.from_json(json_str)
print(type(request).__name__)  # JSONRPC20Request
print(request.method)  # add
print(request.args)    # (1, 2)

# JSON-RPC 1.0 (no "jsonrpc" field)
json_str = '{"method": "add", "params": [1, 2], "id": 1}'  
request = JSONRPCRequest.from_json(json_str)
print(type(request).__name__)  # JSONRPC10Request

# Batch request
batch_json = '''[
    {"jsonrpc": "2.0", "method": "sum", "params": [1,2,4], "id": "1"},
    {"jsonrpc": "2.0", "method": "notify_hello", "params": [7]},
    {"jsonrpc": "2.0", "method": "subtract", "params": [42,23], "id": "2"}
]'''
batch_request = JSONRPCRequest.from_json(batch_json)
print(type(batch_request).__name__)  # JSONRPC20BatchRequest
print(len(batch_request.requests))   # 3

Working with Parameters

# Access parameters in different ways
request = JSONRPC20Request(
    method="calculate", 
    params={"operation": "add", "x": 10, "y": 5}
)

# As keyword arguments
print(request.kwargs)  # {"operation": "add", "x": 10, "y": 5}
print(request.args)    # ()

# Positional parameters
request2 = JSONRPC20Request(method="add", params=[10, 5])
print(request2.args)   # (10, 5)  
print(request2.kwargs) # {}

# Mixed access for method dispatch
def my_method(*args, **kwargs):
    return sum(args) + sum(kwargs.values())

result = my_method(*request2.args, **request2.kwargs)  # 15

Creating Responses

from jsonrpc.jsonrpc2 import JSONRPC20Response
from jsonrpc.exceptions import JSONRPCError

# Success response
response = JSONRPC20Response(result=42, _id=1)
print(response.json)
# {"jsonrpc": "2.0", "result": 42, "id": 1}

# Error response
error = JSONRPCError(code=-32602, message="Invalid params", data={"param": "x"})
response = JSONRPC20Response(error=error._data, _id=1)
print(response.json)
# {"jsonrpc": "2.0", "error": {"code": -32602, "message": "Invalid params", "data": {"param": "x"}}, "id": 1}

# Link response to request
response.request = request

Batch Processing

from jsonrpc.jsonrpc2 import JSONRPC20BatchRequest, JSONRPC20BatchResponse

# Create individual requests
req1 = JSONRPC20Request(method="add", params=[1, 2], _id=1)
req2 = JSONRPC20Request(method="multiply", params=[3, 4], _id=2)  
req3 = JSONRPC20Request(method="notify", params=["hello"], is_notification=True)

# Create batch
batch = JSONRPC20BatchRequest(req1, req2, req3)
print(batch.json)

# Process responses
resp1 = JSONRPC20Response(result=3, _id=1)
resp2 = JSONRPC20Response(result=12, _id=2)
# No response for notification

batch_response = JSONRPC20BatchResponse(resp1, resp2)
print(batch_response.json)
# [{"jsonrpc": "2.0", "result": 3, "id": 1}, {"jsonrpc": "2.0", "result": 12, "id": 2}]

Install with Tessl CLI