class JsonRpcHandler(tornado.web.RequestHandler):
"""Tornado handler that speaks JSON-RPC 2.0 over HTTP POST.
Expects ``self.application.settings['namespaces']`` to be a dict mapping
prefix strings to ``lythonic.compose.namespace.Namespace`` instances.
"""
@override
def prepare(self) -> None:
self.set_header("Content-Type", "application/json")
if not self.application.settings.get("auth_enabled", False):
return
auth_db = self.application.settings.get("auth_db")
if auth_db is None:
return
token = self._extract_bearer_token()
if not token:
self._write_unauthorized()
return
from woodglue.token_store import validate_token
if not validate_token(auth_db, token):
self._write_unauthorized()
return
def _extract_bearer_token(self) -> str:
auth_header = self.request.headers.get("Authorization", "")
if auth_header.startswith("Bearer "):
return auth_header[7:].strip()
return ""
def _write_unauthorized(self) -> None:
self.write(
{"jsonrpc": "2.0", "error": {"code": -32000, "message": "Unauthorized"}, "id": None}
)
self.finish()
@override
async def post(self) -> None:
request_id: Any = None
# Parse JSON body
try:
body = json.loads(self.request.body)
except (json.JSONDecodeError, TypeError):
self.write(_error_response(PARSE_ERROR, "Parse error"))
return
# Batch requests are not supported
if isinstance(body, list):
self.write(_error_response(INVALID_REQUEST, "Batch requests are not supported"))
return
request_id = body.get("id")
# Validate required fields
if not isinstance(body, dict) or body.get("jsonrpc") != "2.0" or "method" not in body:
self.write(
_error_response(
INVALID_REQUEST,
"Invalid Request: missing 'jsonrpc' or 'method'",
request_id,
)
)
return
method: str = body["method"]
params: Any = body.get("params")
# Resolve namespace and method via dot prefix
method_index: dict[str, dict[str, Any]] = self.application.settings["method_index"]
dot_pos = method.find(".")
if dot_pos < 0:
self.write(_error_response(METHOD_NOT_FOUND, f"Method not found: {method}", request_id))
return
prefix = method[:dot_pos]
method_name = method[dot_pos + 1 :]
methods = method_index.get(prefix)
if methods is None:
self.write(_error_response(METHOD_NOT_FOUND, f"Method not found: {method}", request_id))
return
node = methods.get(method_name)
if node is None:
self.write(_error_response(METHOD_NOT_FOUND, f"Method not found: {method}", request_id))
return
# Build kwargs from params
kwargs: dict[str, Any] = {}
method_args = node.method.args
if params is not None:
if isinstance(params, list):
# Positional params: zip with declared arg names
for arg_info, value in zip(method_args, params, strict=False):
kwargs[arg_info.name] = value
elif isinstance(params, dict):
kwargs = dict(params)
else:
self.write(
_error_response(
INVALID_PARAMS,
"params must be an array or object",
request_id,
)
)
return
# Validate required params
for arg_info in method_args:
if not arg_info.is_optional and arg_info.name not in kwargs:
self.write(
_error_response(
INVALID_PARAMS,
f"Missing required parameter: {arg_info.name}",
request_id,
)
)
return
# Deserialize BaseModel params
try:
for arg_info in method_args:
if (
arg_info.name in kwargs
and isinstance(arg_info.annotation, type)
and issubclass(arg_info.annotation, BaseModel)
):
kwargs[arg_info.name] = arg_info.annotation.model_validate(
kwargs[arg_info.name]
)
except ValidationError as exc:
self.write(
_error_response(
INVALID_PARAMS,
f"Invalid parameters: {exc}",
request_id,
)
)
return
# Set current_mount context var for this namespace
from woodglue.mount import MountContext, current_mount
mounts: dict[str, MountContext] = self.application.settings.get("mounts", {})
mount = mounts.get(prefix)
token = current_mount.set(mount) if mount else None
# Call the method
try:
result = node(**kwargs)
if inspect.isawaitable(result):
result = await result
except Exception:
logger.exception("Internal error calling %s", method)
self.write(_error_response(INTERNAL_ERROR, "Internal error", request_id))
return
finally:
if token is not None:
current_mount.reset(token)
# Return result
self.write(
{
"jsonrpc": "2.0",
"result": _serialize_result(result),
"id": request_id,
}
)