json-rpc

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}")