Error payload helpers for streaming provider paths.
normalize_openai_error_payload
normalize_openai_error_payload(payload)
Return a minimal OpenAI-compatible error envelope.
Source code in ccproxy/streaming/errors.py
| def normalize_openai_error_payload(payload: Any) -> dict[str, Any]:
"""Return a minimal OpenAI-compatible error envelope."""
data = _decode_error_payload(payload)
if isinstance(data, dict) and isinstance(data.get("error"), dict):
error = dict(data["error"])
message = error.get("message")
if not isinstance(message, str) or not message:
error["message"] = "upstream error"
error.setdefault("type", "invalid_request_error")
error.setdefault("param", None)
error.setdefault("code", None)
return {"error": error}
message = _extract_error_message(data)
param = _extract_unsupported_parameter(message)
return {
"error": {
"type": "invalid_request_error",
"message": message or "upstream error",
"param": param,
"code": "unsupported_parameter" if param else None,
}
}
|