jsonrpc.py•2.48 kB
"""
JSON-RPC 2.0 message handling for MCP server.
Handles parsing requests and formatting responses.
"""
import json
from typing import Any, Dict, Optional
class JSONRPCError(Exception):
"""Custom exception for JSON-RPC errors."""
def __init__(self, code: int, message: str, data: Any = None):
self.code = code
self.message = message
self.data = data
super().__init__(message)
# JSON-RPC error codes
PARSE_ERROR = -32700
INVALID_REQUEST = -32600
METHOD_NOT_FOUND = -32601
INVALID_PARAMS = -32602
INTERNAL_ERROR = -32603
def parse_request(line: str) -> Dict[str, Any]:
"""
Parse a JSON-RPC request from a line of input.
Args:
line: JSON string containing the request
Returns:
Parsed request dictionary
Raises:
JSONRPCError: If parsing fails or request is invalid
"""
try:
request = json.loads(line)
except json.JSONDecodeError as e:
raise JSONRPCError(PARSE_ERROR, f"Parse error: {e}")
# Validate required fields
if not isinstance(request, dict):
raise JSONRPCError(INVALID_REQUEST, "Request must be an object")
if "jsonrpc" not in request or request["jsonrpc"] != "2.0":
raise JSONRPCError(INVALID_REQUEST, "Missing or invalid jsonrpc version")
if "method" not in request:
raise JSONRPCError(INVALID_REQUEST, "Missing method field")
return request
def create_response(request_id: Any, result: Any) -> str:
"""
Create a JSON-RPC success response.
Args:
request_id: ID from the original request
result: Result data to return
Returns:
JSON string response
"""
response = {
"jsonrpc": "2.0",
"id": request_id,
"result": result
}
return json.dumps(response)
def create_error_response(request_id: Any, code: int, message: str, data: Any = None) -> str:
"""
Create a JSON-RPC error response.
Args:
request_id: ID from the original request
code: Error code
message: Error message
data: Optional additional error data
Returns:
JSON string response
"""
error = {
"code": code,
"message": message
}
if data is not None:
error["data"] = data
response = {
"jsonrpc": "2.0",
"id": request_id,
"error": error
}
return json.dumps(response)