JSON-RPC

1.15.0 · active · verified Thu Apr 16

The `json-rpc` library provides a Python implementation of the JSON-RPC 2.0 protocol, offering tools for both server-side request dispatching and client-side request construction. It focuses on the protocol messaging rather than the underlying transport. The current version is 1.15.0, and it maintains a stable release cadence primarily addressing bug fixes and minor enhancements.

Common errors

Warnings

Install

Imports

Quickstart

This quickstart demonstrates the core functionality of `json-rpc`: how `JSONRPCResponseManager` dispatches incoming JSON-RPC requests on the server side and how `JSONRPCClient` constructs outgoing requests. It includes examples of successful calls, parameter passing, and error handling. Note that the actual network transport (e.g., HTTP POST) needs to be implemented separately, often using libraries like `requests` for clients or `werkzeug`/Flask for servers.

import json
import os
from jsonrpc.manager import JSONRPCResponseManager
from jsonrpc.client import JSONRPCClient

# --- Server-side logic (how to handle an incoming JSON-RPC request) ---

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

def greet(name="Guest"):
    return f"Hello, {name}!"

def secure_data(token):
    # Simulate an authentication check using an environment variable
    expected_token = os.environ.get('AUTH_TOKEN', 'super_secret_token_123')
    if token == expected_token:
        return "Sensitive data accessed."
    else:
        # JSON-RPC error response is automatically handled by the manager
        raise ValueError("Invalid token")

# Define a dispatcher mapping method names to Python functions
dispatcher = {
    "add": add,
    "greet": greet,
    "secure_data": secure_data,
}

# Simulate an incoming raw JSON-RPC request string
incoming_request_str = json.dumps({
    "jsonrpc": "2.0",
    "method": "add",
    "params": {"a": 5, "b": 3},
    "id": "req-123"
})

# Process the request using the manager
print("--- Server processing request ---")
response_object = JSONRPCResponseManager.handle(incoming_request_str, dispatcher)

if response_object:
    # `response_object.json` contains the Python dict representation of the JSON-RPC response
    print(json.dumps(response_object.json, indent=2))
else:
    print("No response object (e.g., for a notification without 'id').")

# Simulate a request with authentication failure
print("\n--- Server processing authenticated request with wrong token ---")
auth_request_str = json.dumps({
    "jsonrpc": "2.0",
    "method": "secure_data",
    "params": ["wrong_token"],
    "id": "auth-req-456"
})
auth_response_object = JSONRPCResponseManager.handle(auth_request_str, dispatcher)
if auth_response_object:
    print(json.dumps(auth_response_object.json, indent=2))

# --- Client-side logic (how to construct a JSON-RPC request) ---

# JSONRPCClient helps in constructing the request payload.
# It does NOT handle the actual network transport (e.g., HTTP POST).
# You would typically subclass it or use its .call() method to get the payload,
# then use a library like 'requests' to send it.

class MyDummyTransportClient(JSONRPCClient):
    # This dummy client just prints the request payload
    # In a real app, this method would send an HTTP POST request
    # and return the response text.
    def _send_request(self, request_str):
        print(f"\n--- Client sending request payload ---\n{request_str}")
        # In a real scenario, you'd send this via HTTP and get a response.
        # For this quickstart, we'll return a simulated server response.
        
        # Simulate the server handling this request
        simulated_response_object = JSONRPCResponseManager.handle(request_str, dispatcher)
        if simulated_response_object:
            return json.dumps(simulated_response_object.json)
        return json.dumps({
            "jsonrpc": "2.0",
            "error": {"code": -32000, "message": "Server error during simulation"},
            "id": None
        })

print("\n--- Client constructing and simulating a call ---")
client = MyDummyTransportClient(service_url="http://example.com/api") # URL is dummy for this example
result = client.call("greet", name="Alice")
print(f"Client received simulated result: {result}")

# Example of a client calling an RPC method that causes an error on the server
print("\n--- Client constructing and simulating an error-causing call ---")
error_result = client.call("secure_data", token="bad_token")
print(f"Client received simulated error result: {error_result}")

view raw JSON →