"""JSON-RPC 2.0 server primitives for I2PControl.""" from __future__ import annotations import json from dataclasses import dataclass, field from typing import Any # Standard JSON-RPC 2.0 error codes PARSE_ERROR = -32700 INVALID_REQUEST = -32600 METHOD_NOT_FOUND = -32601 INTERNAL_ERROR = -32603 @dataclass class JSONRPCRequest: """Parsed JSON-RPC 2.0 request.""" method: str params: dict = field(default_factory=dict) id: int | str | None = None jsonrpc: str = "2.0" @dataclass class JSONRPCResponse: """JSON-RPC 2.0 response.""" result: dict | None = None error: dict | None = None id: int | str | None = None jsonrpc: str = "2.0" def to_dict(self) -> dict[str, Any]: d: dict[str, Any] = {"jsonrpc": self.jsonrpc, "id": self.id} if self.error is not None: d["error"] = self.error else: d["result"] = self.result return d def parse_jsonrpc_request(data: str) -> JSONRPCRequest: """Parse a JSON-RPC 2.0 request string. Raises ``ValueError`` on invalid JSON, missing ``jsonrpc`` field, or missing ``method`` field. """ try: obj = json.loads(data) except (json.JSONDecodeError, TypeError) as exc: raise ValueError("Parse error: invalid JSON") from exc if not isinstance(obj, dict): raise ValueError("Parse error: request must be a JSON object") if "jsonrpc" not in obj or obj["jsonrpc"] != "2.0": raise ValueError("Invalid request: missing or wrong jsonrpc field") if "method" not in obj: raise ValueError("Invalid request: missing method field") return JSONRPCRequest( method=obj["method"], params=obj.get("params", {}), id=obj.get("id"), jsonrpc=obj["jsonrpc"], ) def build_jsonrpc_response( result: dict | None = None, error: dict | None = None, request_id: int | str | None = None, ) -> dict[str, Any]: """Build a JSON-RPC 2.0 response dict.""" resp: dict[str, Any] = {"jsonrpc": "2.0", "id": request_id} if error is not None: resp["error"] = error else: resp["result"] = result return resp def build_jsonrpc_error( code: int, message: str, request_id: int | str | None = None, ) -> dict[str, Any]: """Build a JSON-RPC 2.0 error response.""" return build_jsonrpc_response( error={"code": code, "message": message}, request_id=request_id, )