from typing import Any from fastapi import FastAPI, Request from fastapi.exceptions import RequestValidationError from fastapi.responses import JSONResponse class OpenAICompatibleError(Exception): def __init__( self, *, status_code: int, message: str, error_type: str = "invalid_request_error", param: str | None = None, code: str | None = None, ) -> None: self.status_code = status_code self.message = message self.error_type = error_type self.param = param self.code = code super().__init__(message) def error_payload(message: str, error_type: str, param: str | None = None, code: str | None = None) -> dict[str, Any]: return { "error": { "message": message, "type": error_type, "param": param, "code": code, } } async def openai_error_handler(_: Request, exc: OpenAICompatibleError) -> JSONResponse: return JSONResponse( status_code=exc.status_code, content=error_payload(exc.message, exc.error_type, exc.param, exc.code), ) async def validation_error_handler(_: Request, exc: RequestValidationError) -> JSONResponse: first = exc.errors()[0] if exc.errors() else {} loc = first.get("loc", []) param = loc[-1] if loc else None msg = first.get("msg", "Invalid request body") return JSONResponse( status_code=422, content=error_payload(msg, "invalid_request_error", str(param) if param else None, None), ) async def unexpected_error_handler(_: Request, exc: Exception) -> JSONResponse: return JSONResponse( status_code=500, content=error_payload(str(exc) or "Internal server error", "server_error", None, None), ) def register_exception_handlers(app: FastAPI) -> None: app.add_exception_handler(OpenAICompatibleError, openai_error_handler) app.add_exception_handler(RequestValidationError, validation_error_handler) app.add_exception_handler(Exception, unexpected_error_handler)