Coverage for mcpgateway / validation / jsonrpc.py: 100%

1# -*- coding: utf-8 -*- 

2"""Location: ./mcpgateway/validation/jsonrpc.py 

3Copyright 2025 

4SPDX-License-Identifier: Apache-2.0 

5Authors: Mihai Criveti 

6 

7JSON-RPC Validation. 

8This module provides validation functions for JSON-RPC 2.0 requests and responses 

9according to the specification at https://www.jsonrpc.org/specification. 

10 

11Includes: 

12- Request validation 

13- Response validation 

14- Standard error codes 

15- Error message formatting 

16 

17Examples: 

18 >>> from mcpgateway.validation.jsonrpc import JSONRPCError, validate_request 

19 >>> error = JSONRPCError(-32600, "Invalid Request") 

20 >>> error.code 

21 -32600 

22 >>> error.message 

23 'Invalid Request' 

24 >>> validate_request({'jsonrpc': '2.0', 'method': 'test', 'id': 1}) 

25 >>> validate_request({'jsonrpc': '2.0', 'method': 'test'}) # notification 

26 >>> try: 

27 ... validate_request({'method': 'test'}) # missing jsonrpc 

28 ... except JSONRPCError as e: 

29 ... e.code 

30 -32600 

31""" 

32 

33# Standard 

34from typing import Any, Dict, Optional, Union 

35 

36 

37class JSONRPCError(Exception): 

38 """JSON-RPC protocol error.""" 

39 

40 def __init__( 

41 self, 

42 code: int, 

43 message: str, 

44 data: Optional[Any] = None, 

45 request_id: Optional[Union[str, int]] = None, 

46 ): 

47 """Initialize JSON-RPC error. 

48 

49 Args: 

50 code: Error code 

51 message: Error message 

52 data: Optional error data 

53 request_id: Optional request ID 

54 """ 

55 self.code = code 

56 self.message = message 

57 self.data = data 

58 self.request_id = request_id 

59 super().__init__(message) 

60 

61 def to_dict(self) -> Dict[str, Any]: 

62 """Convert error to JSON-RPC error response dict. 

63 

64 Returns: 

65 Error response dictionary 

66 

67 Examples: 

68 Basic error without data: 

69 >>> error = JSONRPCError(-32600, "Invalid Request", request_id=1) 

70 >>> error.to_dict() 

71 {'jsonrpc': '2.0', 'error': {'code': -32600, 'message': 'Invalid Request'}, 'request_id': 1} 

72 

73 Error with additional data: 

74 >>> error = JSONRPCError(-32602, "Invalid params", data={"param": "value"}, request_id="abc") 

75 >>> error.to_dict() 

76 {'jsonrpc': '2.0', 'error': {'code': -32602, 'message': 'Invalid params', 'data': {'param': 'value'}}, 'request_id': 'abc'} 

77 

78 Error without request ID (for parse errors): 

79 >>> error = JSONRPCError(-32700, "Parse error", data="Unexpected EOF") 

80 >>> error.to_dict() 

81 {'jsonrpc': '2.0', 'error': {'code': -32700, 'message': 'Parse error', 'data': 'Unexpected EOF'}, 'request_id': None} 

82 

83 Error with complex data: 

84 >>> error = JSONRPCError(-32603, "Internal error", data={"details": ["error1", "error2"], "timestamp": 123456}, request_id=42) 

85 >>> sorted(error.to_dict()['error']['data']['details']) 

86 ['error1', 'error2'] 

87 """ 

88 error = {"code": self.code, "message": self.message} 

89 if self.data is not None: 

90 error["data"] = self.data 

91 

92 return {"jsonrpc": "2.0", "error": error, "request_id": self.request_id} 

93 

94 

95# Standard JSON-RPC error codes 

96PARSE_ERROR = -32700 #: Invalid JSON 

97INVALID_REQUEST = -32600 #: Invalid Request object 

98METHOD_NOT_FOUND = -32601 #: Method not found 

99INVALID_PARAMS = -32602 #: Invalid method parameters 

100INTERNAL_ERROR = -32603 #: Internal JSON-RPC error 

101SERVER_ERROR_START = -32000 #: Start of server error codes 

102SERVER_ERROR_END = -32099 #: End of server error codes 

103 

104 

105def validate_request(request: Dict[str, Any]) -> None: 

106 """Validate JSON-RPC request. 

107 

108 Args: 

109 request: Request dictionary to validate 

110 

111 Raises: 

112 JSONRPCError: If request is invalid 

113 

114 Examples: 

115 Valid request: 

116 >>> validate_request({"jsonrpc": "2.0", "method": "ping", "id": 1}) 

117 

118 Valid notification (no id): 

119 >>> validate_request({"jsonrpc": "2.0", "method": "notify"}) 

120 

121 Valid request with params: 

122 >>> validate_request({"jsonrpc": "2.0", "method": "add", "params": [1, 2], "id": 1}) 

123 >>> validate_request({"jsonrpc": "2.0", "method": "add", "params": {"a": 1, "b": 2}, "id": 1}) 

124 

125 Invalid version: 

126 >>> validate_request({"jsonrpc": "1.0", "method": "ping", "id": 1}) # doctest: +ELLIPSIS 

127 Traceback (most recent call last): 

128 ... 

129 mcpgateway.validation.jsonrpc.JSONRPCError: Invalid JSON-RPC version 

130 

131 Missing method: 

132 >>> validate_request({"jsonrpc": "2.0", "id": 1}) # doctest: +ELLIPSIS 

133 Traceback (most recent call last): 

134 ... 

135 mcpgateway.validation.jsonrpc.JSONRPCError: Invalid or missing method 

136 

137 Empty method: 

138 >>> validate_request({"jsonrpc": "2.0", "method": "", "id": 1}) # doctest: +ELLIPSIS 

139 Traceback (most recent call last): 

140 ... 

141 mcpgateway.validation.jsonrpc.JSONRPCError: Invalid or missing method 

142 

143 Invalid params type: 

144 >>> validate_request({"jsonrpc": "2.0", "method": "test", "params": "invalid", "id": 1}) # doctest: +ELLIPSIS 

145 Traceback (most recent call last): 

146 ... 

147 mcpgateway.validation.jsonrpc.JSONRPCError: Invalid params type 

148 

149 Invalid ID type: 

150 >>> validate_request({"jsonrpc": "2.0", "method": "test", "id": True}) # doctest: +ELLIPSIS 

151 Traceback (most recent call last): 

152 ... 

153 mcpgateway.validation.jsonrpc.JSONRPCError: Invalid request ID type 

154 """ # doctest: +ELLIPSIS 

155 # Check jsonrpc version 

156 if request.get("jsonrpc") != "2.0": 

157 raise JSONRPCError(INVALID_REQUEST, "Invalid JSON-RPC version", request_id=request.get("id")) 

158 

159 # Check method 

160 method = request.get("method") 

161 if not isinstance(method, str) or not method: 

162 raise JSONRPCError(INVALID_REQUEST, "Invalid or missing method", request_id=request.get("id")) 

163 

164 # Check ID for requests (not notifications) 

165 if "id" in request: 

166 request_id = request["id"] 

167 if not isinstance(request_id, (str, int)) or isinstance(request_id, bool): 

168 raise JSONRPCError(INVALID_REQUEST, "Invalid request ID type", request_id=None) 

169 

170 # Check params if present 

171 params = request.get("params") 

172 if params is not None: 

173 if not isinstance(params, (dict, list)): 

174 raise JSONRPCError(INVALID_REQUEST, "Invalid params type", request_id=request.get("id")) 

175 

176 

177def validate_response(response: Dict[str, Any]) -> None: 

178 """Validate JSON-RPC response. 

179 

180 Args: 

181 response: Response dictionary to validate 

182 

183 Raises: 

184 JSONRPCError: If response is invalid 

185 

186 Examples: 

187 Valid success response: 

188 >>> validate_response({"jsonrpc": "2.0", "result": 42, "id": 1}) 

189 

190 Valid error response: 

191 >>> validate_response({"jsonrpc": "2.0", "error": {"code": -32601, "message": "Method not found"}, "id": 1}) 

192 

193 Valid response with null result: 

194 >>> validate_response({"jsonrpc": "2.0", "result": None, "id": 1}) 

195 

196 Valid response with null id (for errors during id parsing): 

197 >>> validate_response({"jsonrpc": "2.0", "error": {"code": -32700, "message": "Parse error"}, "id": None}) 

198 

199 Invalid version: 

200 >>> validate_response({"jsonrpc": "1.0", "result": 42, "id": 1}) # doctest: +ELLIPSIS 

201 Traceback (most recent call last): 

202 ... 

203 mcpgateway.validation.jsonrpc.JSONRPCError: Invalid JSON-RPC version 

204 

205 Missing ID: 

206 >>> validate_response({"jsonrpc": "2.0", "result": 42}) # doctest: +ELLIPSIS 

207 Traceback (most recent call last): 

208 ... 

209 mcpgateway.validation.jsonrpc.JSONRPCError: Missing response ID 

210 

211 Invalid ID type (boolean): 

212 >>> validate_response({"jsonrpc": "2.0", "result": 42, "id": True}) # doctest: +ELLIPSIS 

213 Traceback (most recent call last): 

214 ... 

215 mcpgateway.validation.jsonrpc.JSONRPCError: Invalid response ID type 

216 

217 Invalid ID type (list): 

218 >>> validate_response({"jsonrpc": "2.0", "result": 42, "id": [1, 2]}) # doctest: +ELLIPSIS 

219 Traceback (most recent call last): 

220 ... 

221 mcpgateway.validation.jsonrpc.JSONRPCError: Invalid response ID type 

222 

223 Missing both result and error: 

224 >>> validate_response({"jsonrpc": "2.0", "id": 1}) # doctest: +ELLIPSIS 

225 Traceback (most recent call last): 

226 ... 

227 mcpgateway.validation.jsonrpc.JSONRPCError: Response must contain either result or error 

228 

229 Both result and error present: 

230 >>> validate_response({"jsonrpc": "2.0", "result": 42, "error": {"code": -1, "message": "Error"}, "id": 1}) # doctest: +ELLIPSIS 

231 Traceback (most recent call last): 

232 ... 

233 mcpgateway.validation.jsonrpc.JSONRPCError: Response cannot contain both result and error 

234 

235 Invalid error object type: 

236 >>> validate_response({"jsonrpc": "2.0", "error": "Invalid error", "id": 1}) # doctest: +ELLIPSIS 

237 Traceback (most recent call last): 

238 ... 

239 mcpgateway.validation.jsonrpc.JSONRPCError: Invalid error object type 

240 

241 Error missing code: 

242 >>> validate_response({"jsonrpc": "2.0", "error": {"message": "Error"}, "id": 1}) # doctest: +ELLIPSIS 

243 Traceback (most recent call last): 

244 ... 

245 mcpgateway.validation.jsonrpc.JSONRPCError: Error must contain code and message 

246 

247 Error missing message: 

248 >>> validate_response({"jsonrpc": "2.0", "error": {"code": -32601}, "id": 1}) # doctest: +ELLIPSIS 

249 Traceback (most recent call last): 

250 ... 

251 mcpgateway.validation.jsonrpc.JSONRPCError: Error must contain code and message 

252 

253 Invalid error code type: 

254 >>> validate_response({"jsonrpc": "2.0", "error": {"code": "invalid", "message": "Error"}, "id": 1}) # doctest: +ELLIPSIS 

255 Traceback (most recent call last): 

256 ... 

257 mcpgateway.validation.jsonrpc.JSONRPCError: Error code must be integer 

258 

259 Invalid error message type: 

260 >>> validate_response({"jsonrpc": "2.0", "error": {"code": -32601, "message": 123}, "id": 1}) # doctest: +ELLIPSIS 

261 Traceback (most recent call last): 

262 ... 

263 mcpgateway.validation.jsonrpc.JSONRPCError: Error message must be string 

264 

265 Valid error with additional data: 

266 >>> validate_response({"jsonrpc": "2.0", "error": {"code": -32602, "message": "Invalid params", "data": {"param": "name"}}, "id": 1}) 

267 """ 

268 # Check jsonrpc version 

269 if response.get("jsonrpc") != "2.0": 

270 raise JSONRPCError(INVALID_REQUEST, "Invalid JSON-RPC version", request_id=response.get("id")) 

271 

272 # Check ID 

273 if "id" not in response: 

274 raise JSONRPCError(INVALID_REQUEST, "Missing response ID", request_id=None) 

275 

276 response_id = response["id"] 

277 if not isinstance(response_id, (str, int, type(None))) or isinstance(response_id, bool): 

278 raise JSONRPCError(INVALID_REQUEST, "Invalid response ID type", request_id=None) 

279 

280 # Check result XOR error 

281 has_result = "result" in response 

282 has_error = "error" in response 

283 

284 if not has_result and not has_error: 

285 raise JSONRPCError(INVALID_REQUEST, "Response must contain either result or error", request_id=id) 

286 if has_result and has_error: 

287 raise JSONRPCError(INVALID_REQUEST, "Response cannot contain both result and error", request_id=id) 

288 

289 # Validate error object 

290 if has_error: 

291 error = response["error"] 

292 if not isinstance(error, dict): 

293 raise JSONRPCError(INVALID_REQUEST, "Invalid error object type", request_id=id) 

294 

295 if "code" not in error or "message" not in error: 

296 raise JSONRPCError(INVALID_REQUEST, "Error must contain code and message", request_id=id) 

297 

298 if not isinstance(error["code"], int): 

299 raise JSONRPCError(INVALID_REQUEST, "Error code must be integer", request_id=id) 

300 

301 if not isinstance(error["message"], str): 

302 raise JSONRPCError(INVALID_REQUEST, "Error message must be string", request_id=id)