### Install jsonrpcclient Source: https://github.com/explodinglabs/jsonrpcclient/blob/main/README.md Use pip to install the library in your Python environment. ```sh pip install jsonrpcclient ``` -------------------------------- ### Basic JSON-RPC Request with Requests Source: https://github.com/explodinglabs/jsonrpcclient/wiki/Requests Use this snippet to send a single JSON-RPC request and parse the response. Ensure the requests library is installed. ```python import logging import requests from jsonrpcclient import Error, Ok, parse, request response = requests.post("http://localhost:5000/", json=request("ping"), timeout=10) parsed = parse(response.json()) if isinstance(parsed, Ok): print(parsed.result) elif isinstance(parsed, Error): logging.error(parsed.message) ``` -------------------------------- ### Python 3.10 Match Statement for JSON-RPC Response Source: https://github.com/explodinglabs/jsonrpcclient/wiki/Requests This example shows how to handle JSON-RPC responses using Python 3.10's match statement for cleaner conditional logic. Requires Python 3.10 or later. ```python import logging import requests from jsonrpcclient import Error, Ok, parse, request response = requests.post("http://localhost:5000/", json=request("ping"), timeout=10) match parse(response.json()): case Ok(result, id_): print(result) case Error(code, message, data, id_): logging.error(message) ``` -------------------------------- ### Create JSON-RPC Requests Source: https://context7.com/explodinglabs/jsonrpcclient/llms.txt Use request() to generate a request dictionary with an auto-incrementing ID, or request_json() to get a JSON string. ```python from jsonrpcclient import request, request_json # Basic request with auto-incrementing ID req = request("ping") # => {'jsonrpc': '2.0', 'method': 'ping', 'id': 1} # Request with positional parameters req = request("add", params=[2, 3], id=1) # => {'jsonrpc': '2.0', 'method': 'add', 'params': [2, 3], 'id': 1} # Request with named parameters req = request("user.get", params={"id": 42, "include_email": True}, id=1) # => {'jsonrpc': '2.0', 'method': 'user.get', 'params': {'id': 42, 'include_email': True}, 'id': 1} # Get JSON string directly json_req = request_json("ping", id=1) # => '{"jsonrpc": "2.0", "method": "ping", "id": 1}' ``` -------------------------------- ### Custom Decimal ID Generator Source: https://context7.com/explodinglabs/jsonrpcclient/llms.txt Generates sequential decimal request IDs starting from a specified number. Use `next(gen)` to get the next ID. ```python from jsonrpcclient import id_generators # Decimal starting from 100 gen = id_generators.decimal(start=100) print(next(gen)) # => 100 print(next(gen)) # => 101 ``` -------------------------------- ### Custom Hexadecimal ID Generator Source: https://context7.com/explodinglabs/jsonrpcclient/llms.txt Generates sequential hexadecimal request IDs starting from a specified number. Use `next(gen)` to get the next ID. ```python from jsonrpcclient import id_generators # Hexadecimal starting from 255 gen = id_generators.hexadecimal(start=255) print(next(gen)) # => 'ff' print(next(gen)) # => '100' ``` -------------------------------- ### Integration with WebSockets Source: https://context7.com/explodinglabs/jsonrpcclient/llms.txt Handle real-time bidirectional communication using JSON string variants with WebSocket libraries. ```python import asyncio import logging from websockets.client import connect from jsonrpcclient import Error, Ok, parse_json, request_json async def websocket_rpc(): async with connect("ws://localhost:5000/ws") as socket: # Send JSON-RPC request as string await socket.send(request_json("subscribe", params={"channel": "updates"})) # Receive and parse response response = await socket.recv() parsed = parse_json(response) if isinstance(parsed, Ok): print(f"Subscribed: {parsed.result}") elif isinstance(parsed, Error): logging.error(f"Subscription failed: {parsed.message}") asyncio.run(websocket_rpc()) ``` -------------------------------- ### Use Alternative ID Generation Strategies Source: https://context7.com/explodinglabs/jsonrpcclient/llms.txt Generate IDs using hexadecimal, random strings, or UUIDs for specific system requirements. ```python from jsonrpcclient import request_hex, request_random, request_uuid from jsonrpcclient.requests import request_json_hex, request_json_random, request_json_uuid # Hexadecimal IDs (1, 2, ... 9, a, b, c, ...) req = request_hex("ping") # => {'jsonrpc': '2.0', 'method': 'ping', 'id': '1'} # Random 8-character alphanumeric string IDs req = request_random("ping") # => {'jsonrpc': '2.0', 'method': 'ping', 'id': 'fubui5e6'} # UUID IDs for distributed systems req = request_uuid("ping") # => {'jsonrpc': '2.0', 'method': 'ping', 'id': '9bfe2c93-717e-4a45-b91b-55422c5af4ff'} # JSON string versions json_req = request_json_uuid("user.create", params={"name": "Alice"}) # => '{"jsonrpc": "2.0", "method": "user.create", "params": {"name": "Alice"}, "id": "a1b2c3d4-..."}' ``` -------------------------------- ### Structural Pattern Matching Source: https://context7.com/explodinglabs/jsonrpcclient/llms.txt Use Python 3.10+ structural pattern matching to handle Ok and Error responses cleanly. ```python from jsonrpcclient import parse, Ok, Error response = {"jsonrpc": "2.0", "result": {"status": "active", "balance": 150.50}, "id": 1} parsed = parse(response) match parsed: case Ok(result, id_): print(f"Request {id_} succeeded: {result}") case Error(code, message, data, id_): print(f"Request {id_} failed with code {code}: {message}") if data: print(f"Additional info: {data}") ``` -------------------------------- ### Perform Asynchronous JSON-RPC Request with aiohttp Source: https://github.com/explodinglabs/jsonrpcclient/wiki/Aiohttp Uses ClientSession to send a POST request to a JSON-RPC server and parses the response. ```python import asyncio import logging from aiohttp import ClientSession from jsonrpcclient import Error, Ok, parse, request async def main() -> None: """Handle async request""" async with ClientSession() as session: async with session.post( "http://localhost:5000", json=request("ping") ) as response: parsed = parse(await response.json()) # In Python 3.10+ use `match` syntax here instead of isinstance if isinstance(parsed, Ok): print(parsed.result) elif isinstance(parsed, Error): logging.error(parsed.message) asyncio.get_event_loop().run_until_complete(main()) ``` -------------------------------- ### Integration with ZeroMQ Source: https://context7.com/explodinglabs/jsonrpcclient/llms.txt Use ZeroMQ sockets for high-performance messaging with JSON string functions. ```python import logging import zmq from jsonrpcclient import Ok, Error, parse_json, request_json # Create ZeroMQ REQ socket context = zmq.Context() socket = context.socket(zmq.REQ) socket.connect("tcp://localhost:5555") ``` -------------------------------- ### Integration with Requests Library Source: https://context7.com/explodinglabs/jsonrpcclient/llms.txt Perform synchronous JSON-RPC calls over HTTP using the requests library. ```python import logging import requests from jsonrpcclient import Error, Ok, parse, request # Single request response = requests.post( "http://localhost:5000/jsonrpc", json=request("math.add", params=[5, 3]), timeout=10 ) parsed = parse(response.json()) if isinstance(parsed, Ok): print(f"Result: {parsed.result}") # => Result: 8 elif isinstance(parsed, Error): logging.error(f"Error {parsed.code}: {parsed.message}") # Batch request batch = [ request("math.add", params=[1, 2]), request("math.multiply", params=[3, 4]), request("math.divide", params=[10, 2]) ] response = requests.post("http://localhost:5000/jsonrpc", json=batch, timeout=10) for result in parse(response.json()): if isinstance(result, Ok): print(f"ID {result.id}: {result.result}") ``` -------------------------------- ### Integration with aiohttp Source: https://context7.com/explodinglabs/jsonrpcclient/llms.txt Perform non-blocking JSON-RPC calls using aiohttp for asynchronous applications. ```python import asyncio import logging from aiohttp import ClientSession from jsonrpcclient import Error, Ok, parse, request async def call_rpc(method: str, params=None): async with ClientSession() as session: async with session.post( "http://localhost:5000/jsonrpc", json=request(method, params=params) ) as response: return parse(await response.json()) async def main(): # Make async JSON-RPC call parsed = await call_rpc("user.get", {"id": 123}) match parsed: case Ok(result, _): print(f"User: {result['name']}") case Error(code, message, _, _): logging.error(f"Failed: {message}") asyncio.run(main()) ``` -------------------------------- ### Create JSON-RPC Notifications Source: https://context7.com/explodinglabs/jsonrpcclient/llms.txt Use notification() to create fire-and-forget requests that do not include an ID. ```python from jsonrpcclient import notification, notification_json # Simple notification notif = notification("log.debug") # => {'jsonrpc': '2.0', 'method': 'log.debug'} # Notification with positional parameters notif = notification("event.track", params=("click", "button_signup")) # => {'jsonrpc': '2.0', 'method': 'event.track', 'params': ('click', 'button_signup')} # Notification with named parameters notif = notification("metrics.record", params={"cpu": 45.2, "memory": 1024}) # => {'jsonrpc': '2.0', 'method': 'metrics.record', 'params': {'cpu': 45.2, 'memory': 1024}} # Get JSON string directly json_notif = notification_json("heartbeat") # => '{"jsonrpc": "2.0", "method": "heartbeat"}' ``` -------------------------------- ### Generate a JSON-RPC request Source: https://github.com/explodinglabs/jsonrpcclient/blob/main/README.md Create a standard JSON-RPC request dictionary using the request function. ```python from jsonrpcclient import request, parse req = request("ping") # => {'jsonrpc': '2.0', 'method': 'ping', 'id': 1} ``` -------------------------------- ### Send JSON-RPC Request and Handle Response Source: https://context7.com/explodinglabs/jsonrpcclient/llms.txt Demonstrates sending a request to a worker process and handling both successful and error responses. Ensure the socket is properly connected and the context is terminated. ```python socket.send_string(request_json("worker.process", params={"task_id": 42})) response = parse_json(socket.recv().decode()) if isinstance(response, Ok): print(f"Task completed: {response.result}") else: logging.error(f"Task failed: {response.message}") socket.close() context.term() ``` -------------------------------- ### Batch JSON-RPC Requests with Requests Source: https://github.com/explodinglabs/jsonrpcclient/wiki/Requests Send multiple JSON-RPC requests in a single batch. This is useful for improving performance by reducing network overhead. The response will be a list of parsed results. ```python import logging import requests from jsonrpcclient import Error, Ok, parse, request response = requests.post( "http://localhost:5000/", json=[request("ping") for _ in range(5)], timeout=10 ) parsed = parse(response.json()) for p in parsed: if isinstance(p, Ok): print(p.result) elif isinstance(p, Error): logging.error(p.message) ``` -------------------------------- ### Creating Notifications Source: https://context7.com/explodinglabs/jsonrpcclient/llms.txt Functions to create fire-and-forget JSON-RPC requests without an ID. ```APIDOC ## notification(method, params=None) ### Description Creates a JSON-RPC 2.0 notification request, which does not expect a response from the server. ### Parameters - **method** (str) - Required - The name of the RPC method to call. - **params** (list/dict) - Optional - Positional or named parameters for the method. ### Request Example notification("event.track", params=("click", "button_signup")) ### Response - **jsonrpc** (str) - Version string "2.0" - **method** (str) - The method name - **params** (list/dict) - The parameters provided ``` -------------------------------- ### Send JSON-RPC request via ZeroMQ Source: https://github.com/explodinglabs/jsonrpcclient/wiki/ZeroMQ Uses a ZeroMQ REQ socket to send a ping request and handle the response using the jsonrpcclient library. ```python import logging import zmq from jsonrpcclient import Ok, parse_json, request_json socket = zmq.Context().socket(zmq.REQ) socket.connect("tcp://localhost:5000") socket.send_string(request_json("ping")) response = parse_json(socket.recv().decode()) # Python 3.10+ should use `match` syntax here if isinstance(response, Ok): print(response.result) else: logging.error(response.message) ``` -------------------------------- ### Creating JSON-RPC Requests Source: https://context7.com/explodinglabs/jsonrpcclient/llms.txt Functions to generate JSON-RPC 2.0 request objects with automatic ID generation. ```APIDOC ## request(method, params=None, id=None) ### Description Creates a JSON-RPC 2.0 request dictionary with automatic ID generation. ### Parameters - **method** (str) - Required - The name of the RPC method to call. - **params** (list/dict) - Optional - Positional or named parameters for the method. - **id** (int/str) - Optional - Explicit ID for the request. Defaults to auto-incrementing integers. ### Request Example request("add", params=[2, 3], id=1) ### Response - **jsonrpc** (str) - Version string "2.0" - **method** (str) - The method name - **params** (list/dict) - The parameters provided - **id** (int/str) - The request ID ``` -------------------------------- ### Python Websockets JSON-RPC Client Source: https://github.com/explodinglabs/jsonrpcclient/wiki/Websockets Connect to a websocket server, send a JSON-RPC request, and handle the response. Ensure the server is running on ws://localhost:5000. ```python import asyncio import logging from websockets.client import connect from jsonrpcclient import Error, Ok, parse_json, request_json async def main() -> None: """Handle request""" async with connect("ws://localhost:5000") as socket: await socket.send(request_json("ping")) response = parse_json(await socket.recv()) if isinstance(response, Ok): print(response.result) elif isinstance(response, Error): logging.error(response.message) asyncio.get_event_loop().run_until_complete(main()) ``` -------------------------------- ### Parse Successful Responses Source: https://context7.com/explodinglabs/jsonrpcclient/llms.txt Convert response dictionaries or JSON strings into Ok namedtuples for easy result access. ```python from jsonrpcclient import parse, parse_json, Ok # Parse a successful response (dictionary) response = {"jsonrpc": "2.0", "result": "pong", "id": 1} parsed = parse(response) # => Ok(result='pong', id=1) # Access fields print(parsed.result) # => 'pong' print(parsed.id) # => 1 # Parse from JSON string parsed = parse_json('{"jsonrpc": "2.0", "result": {"user": "Alice", "age": 30}, "id": 5}') # => Ok(result={'user': 'Alice', 'age': 30}, id=5) # Check response type if isinstance(parsed, Ok): print(f"Success: {parsed.result}") ``` -------------------------------- ### Parsing Responses Source: https://context7.com/explodinglabs/jsonrpcclient/llms.txt Functions to parse server responses into typed Ok or Error namedtuples. ```APIDOC ## parse(response) ### Description Converts a JSON-RPC response dictionary into a typed namedtuple. ### Parameters - **response** (dict) - Required - The JSON-RPC response object. ### Response - **Ok** (namedtuple) - Returned on success, contains 'result' and 'id'. - **Error** (namedtuple) - Returned on failure, contains 'code', 'message', 'data', and 'id'. ``` -------------------------------- ### Use Custom ID Generator with Request Source: https://context7.com/explodinglabs/jsonrpcclient/llms.txt Integrates a custom ID generator, such as a UUID generator, with the `request_impure` function to create a JSON-RPC request. ```python from jsonrpcclient import id_generators from jsonrpcclient.requests import request_impure # Use custom generator with request_impure custom_gen = id_generators.uuid() req = request_impure(custom_gen, "my.method", params={"key": "value"}) # => {'jsonrpc': '2.0', 'method': 'my.method', 'params': {'key': 'value'}, 'id': 'uuid-here'} ``` -------------------------------- ### Parsing Batch Responses Source: https://context7.com/explodinglabs/jsonrpcclient/llms.txt Process a list of JSON-RPC responses by iterating through the iterator returned by parse(). ```python from jsonrpcclient import parse, Ok, Error # Parse batch response batch_response = [ {"jsonrpc": "2.0", "result": "pong", "id": 1}, {"jsonrpc": "2.0", "error": {"code": -32601, "message": "Method not found"}, "id": 2}, {"jsonrpc": "2.0", "result": 42, "id": 3} ] parsed = parse(batch_response) # Iterate through responses for response in parsed: if isinstance(response, Ok): print(f"ID {response.id}: Success - {response.result}") elif isinstance(response, Error): print(f"ID {response.id}: Error {response.code} - {response.message}") ``` -------------------------------- ### Parse Error Responses Source: https://context7.com/explodinglabs/jsonrpcclient/llms.txt Handle server errors by parsing responses into Error namedtuples containing code, message, and optional data. ```python from jsonrpcclient import parse, parse_json, Error # Parse an error response response = { "jsonrpc": "2.0", "error": {"code": -32601, "message": "Method not found", "data": "unknown_method"}, "id": 1 } parsed = parse(response) # => Error(code=-32601, message='Method not found', data='unknown_method', id=1) ``` -------------------------------- ### Custom Random ID Generator Source: https://context7.com/explodinglabs/jsonrpcclient/llms.txt Generates random request IDs with a specified length and character set. Useful for non-sequential or security-sensitive IDs. ```python from jsonrpcclient import id_generators # Random with custom length and characters gen = id_generators.random(length=12, chars="ABCDEF0123456789") print(next(gen)) # => 'A3F2B1C4D5E6' (12 hex chars) ``` -------------------------------- ### Accessing JSON-RPC Error Fields Source: https://context7.com/explodinglabs/jsonrpcclient/llms.txt Access specific error attributes like code, message, data, and id from a parsed error object. ```python print(parsed.code) # => -32601 print(parsed.message) # => 'Method not found' print(parsed.data) # => 'unknown_method' print(parsed.id) # => 1 # Error without data field parsed = parse_json('{"jsonrpc": "2.0", "error": {"code": -32600, "message": "Invalid Request"}, "id": null}') # => Error(code=-32600, message='Invalid Request', data=None, id=None) ``` -------------------------------- ### Parse a JSON-RPC response Source: https://github.com/explodinglabs/jsonrpcclient/blob/main/README.md Convert a JSON-RPC response dictionary into a result object using the parse function. ```python parsed = parse({"jsonrpc": "2.0", "result": "pong", "id": 1}) # => Ok(result='pong', id=1) ``` === COMPLETE CONTENT === This response contains all available snippets from this library. No additional content exists. Do not make further requests.