feat(V2): error envelope module estratto da server.py
This commit is contained in:
@@ -0,0 +1,51 @@
|
||||
"""Error envelope standard per tutti i tool MCP."""
|
||||
from __future__ import annotations
|
||||
|
||||
import uuid
|
||||
from datetime import UTC, datetime
|
||||
from typing import Any
|
||||
|
||||
|
||||
def error_envelope(
|
||||
*,
|
||||
type_: str,
|
||||
code: str,
|
||||
message: str,
|
||||
retryable: bool,
|
||||
suggested_fix: str | None = None,
|
||||
details: dict | None = None,
|
||||
request_id: str | None = None,
|
||||
) -> dict:
|
||||
env: dict[str, Any] = {
|
||||
"error": {
|
||||
"type": type_,
|
||||
"code": code,
|
||||
"message": message,
|
||||
"retryable": retryable,
|
||||
},
|
||||
"request_id": request_id or uuid.uuid4().hex,
|
||||
"data_timestamp": datetime.now(UTC).isoformat(),
|
||||
}
|
||||
if suggested_fix:
|
||||
env["error"]["suggested_fix"] = suggested_fix
|
||||
if details:
|
||||
env["error"]["details"] = details
|
||||
return env
|
||||
|
||||
|
||||
HTTP_CODE_MAP = {
|
||||
400: "BAD_REQUEST",
|
||||
401: "UNAUTHORIZED",
|
||||
403: "FORBIDDEN",
|
||||
404: "NOT_FOUND",
|
||||
408: "TIMEOUT",
|
||||
409: "CONFLICT",
|
||||
422: "VALIDATION_ERROR",
|
||||
429: "RATE_LIMIT",
|
||||
500: "INTERNAL_ERROR",
|
||||
502: "UPSTREAM_ERROR",
|
||||
503: "UNAVAILABLE",
|
||||
504: "GATEWAY_TIMEOUT",
|
||||
}
|
||||
|
||||
RETRYABLE_STATUSES = frozenset({408, 429, 502, 503, 504})
|
||||
@@ -0,0 +1,29 @@
|
||||
from __future__ import annotations
|
||||
|
||||
from cerbero_mcp.common.errors import error_envelope, HTTP_CODE_MAP
|
||||
|
||||
|
||||
def test_envelope_minimal():
|
||||
e = error_envelope(type_="x", code="C", message="m", retryable=False)
|
||||
assert e["error"]["code"] == "C"
|
||||
assert e["error"]["retryable"] is False
|
||||
assert "request_id" in e
|
||||
assert "data_timestamp" in e
|
||||
|
||||
|
||||
def test_envelope_with_suggested_fix_and_details():
|
||||
e = error_envelope(
|
||||
type_="validation_error",
|
||||
code="INVALID_INPUT",
|
||||
message="bad",
|
||||
retryable=False,
|
||||
suggested_fix="check field x",
|
||||
details={"field": "x"},
|
||||
)
|
||||
assert e["error"]["suggested_fix"] == "check field x"
|
||||
assert e["error"]["details"] == {"field": "x"}
|
||||
|
||||
|
||||
def test_http_code_map_has_common_codes():
|
||||
assert HTTP_CODE_MAP[401] == "UNAUTHORIZED"
|
||||
assert HTTP_CODE_MAP[502] == "UPSTREAM_ERROR"
|
||||
Reference in New Issue
Block a user