Refactor REST error handling to use google.rpc.Status (AIP-193)

Author: knapg

src/a2a/client/transports/rest.py

@@ -34,16 +34,12 @@
     Task,
     TaskPushNotificationConfig,
 )
-from a2a.utils.errors import JSON_RPC_ERROR_CODE_MAP, MethodNotFoundError
+from a2a.utils.errors import A2A_REASON_TO_ERROR, MethodNotFoundError
 from a2a.utils.telemetry import SpanKind, trace_class
 
 
 logger = logging.getLogger(__name__)
 
-_A2A_ERROR_NAME_TO_CLS = {
-    error_type.__name__: error_type for error_type in JSON_RPC_ERROR_CODE_MAP
-}
-
 
 @trace_class(kind=SpanKind.CLIENT)
 class RestTransport(ClientTransport):
@@ -297,15 +293,36 @@ def _get_path(self, base_path: str, tenant: str) -> str:
     def _handle_http_error(self, e: httpx.HTTPStatusError) -> NoReturn:
         """Handles HTTP status errors and raises the appropriate A2AError."""
         try:
-            error_data = e.response.json()
-            error_type = error_data.get('type')
-            message = error_data.get('message', str(e))
+            error_payload = e.response.json()
+            error_data = error_payload.get('error', {})
 
-            if isinstance(error_type, str):
-                # TODO(#723): Resolving imports by name is temporary until proper error handling structure is added in #723.
-                exception_cls = _A2A_ERROR_NAME_TO_CLS.get(error_type)
+            message = error_data.get('message', str(e))
+            details = error_data.get('details', [])
+            if not isinstance(details, list):
+                details = []
+
+            # The `details` array can contain multiple different error objects.
+            # We extract the first `ErrorInfo` object because it contains the
+            # specific `reason` code needed to map this back to a Python A2AError.
+            error_info = {}
+            for d in details:
+                if (
+                    isinstance(d, dict)
+                    and d.get('@type')
+                    == 'type.googleapis.com/google.rpc.ErrorInfo'
+                ):
+                    error_info = d
+                    break
+            reason = error_info.get('reason')
+            metadata = error_info.get('metadata') or {}
+
+            if isinstance(reason, str):
+                exception_cls = A2A_REASON_TO_ERROR.get(reason)
                 if exception_cls:
-                    raise exception_cls(message) from e
+                    exc = exception_cls(message)
+                    if metadata:
+                        exc.data = metadata
+                    raise exc from e
         except (json.JSONDecodeError, ValueError):
             pass
 

src/a2a/server/apps/rest/fastapi_app.py

@@ -7,19 +7,22 @@
 if TYPE_CHECKING:
     from fastapi import APIRouter, FastAPI, Request, Response
     from fastapi.responses import JSONResponse
+    from starlette.exceptions import HTTPException as StarletteHTTPException
 
     _package_fastapi_installed = True
 else:
     try:
         from fastapi import APIRouter, FastAPI, Request, Response
         from fastapi.responses import JSONResponse
+        from starlette.exceptions import HTTPException as StarletteHTTPException
 
         _package_fastapi_installed = True
     except ImportError:
         APIRouter = Any
         FastAPI = Any
         Request = Any
         Response = Any
+        StarletteHTTPException = Any
 
         _package_fastapi_installed = False
 
@@ -36,6 +39,23 @@
 logger = logging.getLogger(__name__)
 
 
+_HTTP_TO_GRPC_STATUS_MAP = {
+    400: 'INVALID_ARGUMENT',
+    401: 'UNAUTHENTICATED',
+    403: 'PERMISSION_DENIED',
+    404: 'NOT_FOUND',
+    405: 'UNIMPLEMENTED',
+    409: 'ALREADY_EXISTS',
+    415: 'INVALID_ARGUMENT',
+    422: 'INVALID_ARGUMENT',
+    500: 'INTERNAL',
+    501: 'UNIMPLEMENTED',
+    502: 'INTERNAL',
+    503: 'UNAVAILABLE',
+    504: 'DEADLINE_EXCEEDED',
+}
+
+
 class A2ARESTFastAPIApplication:
     """A FastAPI application implementing the A2A protocol server REST endpoints.
 
@@ -121,6 +141,34 @@ def build(
             A configured FastAPI application instance.
         """
         app = FastAPI(**kwargs)
+
+        @app.exception_handler(StarletteHTTPException)
+        async def http_exception_handler(
+            request: Request, exc: StarletteHTTPException
+        ) -> Response:
+            """Catches framework-level HTTP exceptions.
+
+            For example, 404 Not Found for bad routes, 422 Unprocessable Entity
+            for schema validation, and formats them into the A2A standard
+            google.rpc.Status JSON format (AIP-193).
+            """
+            grpc_status = _HTTP_TO_GRPC_STATUS_MAP.get(
+                exc.status_code, 'UNKNOWN'
+            )
+            return JSONResponse(
+                status_code=exc.status_code,
+                content={
+                    'error': {
+                        'code': exc.status_code,
+                        'status': grpc_status,
+                        'message': str(exc.detail)
+                        if hasattr(exc, 'detail')
+                        else 'HTTP Exception',
+                    }
+                },
+                media_type='application/json',
+            )
+
         if self.enable_v0_3_compat and self._v03_adapter:
             v03_adapter = self._v03_adapter
             v03_router = APIRouter()

src/a2a/utils/error_handlers.py

@@ -2,7 +2,7 @@
 import logging
 
 from collections.abc import Awaitable, Callable, Coroutine
-from typing import TYPE_CHECKING, Any, cast
+from typing import TYPE_CHECKING, Any
 
 
 if TYPE_CHECKING:
@@ -17,71 +17,15 @@
 
 from google.protobuf.json_format import ParseError
 
-from a2a.server.jsonrpc_models import (
-    InternalError as JSONRPCInternalError,
-)
-from a2a.server.jsonrpc_models import (
-    JSONParseError,
-    JSONRPCError,
-)
 from a2a.utils.errors import (
+    A2A_REST_ERROR_MAPPING,
     A2AError,
-    ContentTypeNotSupportedError,
-    ExtendedAgentCardNotConfiguredError,
-    ExtensionSupportRequiredError,
     InternalError,
-    InvalidAgentResponseError,
-    InvalidParamsError,
-    InvalidRequestError,
-    MethodNotFoundError,
-    PushNotificationNotSupportedError,
-    TaskNotCancelableError,
-    TaskNotFoundError,
-    UnsupportedOperationError,
-    VersionNotSupportedError,
 )
 
 
 logger = logging.getLogger(__name__)
 
-_A2AErrorType = (
-    type[JSONRPCError]
-    | type[JSONParseError]
-    | type[InvalidRequestError]
-    | type[MethodNotFoundError]
-    | type[InvalidParamsError]
-    | type[InternalError]
-    | type[JSONRPCInternalError]
-    | type[TaskNotFoundError]
-    | type[TaskNotCancelableError]
-    | type[PushNotificationNotSupportedError]
-    | type[UnsupportedOperationError]
-    | type[ContentTypeNotSupportedError]
-    | type[InvalidAgentResponseError]
-    | type[ExtendedAgentCardNotConfiguredError]
-    | type[ExtensionSupportRequiredError]
-    | type[VersionNotSupportedError]
-)
-
-A2AErrorToHttpStatus: dict[_A2AErrorType, int] = {
-    JSONRPCError: 500,
-    JSONParseError: 400,
-    InvalidRequestError: 400,
-    MethodNotFoundError: 404,
-    InvalidParamsError: 422,
-    InternalError: 500,
-    JSONRPCInternalError: 500,
-    TaskNotFoundError: 404,
-    TaskNotCancelableError: 409,
-    PushNotificationNotSupportedError: 501,
-    UnsupportedOperationError: 501,
-    ContentTypeNotSupportedError: 415,
-    InvalidAgentResponseError: 502,
-    ExtendedAgentCardNotConfiguredError: 400,
-    ExtensionSupportRequiredError: 400,
-    VersionNotSupportedError: 400,
-}
-
 
 def rest_error_handler(
     func: Callable[..., Awaitable[Response]],
@@ -93,8 +37,8 @@ async def wrapper(*args: Any, **kwargs: Any) -> Response:
         try:
             return await func(*args, **kwargs)
         except A2AError as error:
-            http_code = A2AErrorToHttpStatus.get(
-                cast('_A2AErrorType', type(error)), 500
+            http_code, grpc_status, reason = A2A_REST_ERROR_MAPPING.get(
+                type(error), (500, 'INTERNAL', 'INTERNAL_ERROR')
             )
 
             log_level = (
@@ -107,32 +51,64 @@ async def wrapper(*args: Any, **kwargs: Any) -> Response:
                 "Request error: Code=%s, Message='%s'%s",
                 getattr(error, 'code', 'N/A'),
                 getattr(error, 'message', str(error)),
-                ', Data=' + str(getattr(error, 'data', ''))
-                if getattr(error, 'data', None)
-                else '',
+                f', Data={error.data}' if hasattr(error, 'data') else '',
             )
-            # TODO(#722): Standardize error response format.
+
+            # SECURITY WARNING: Data attached to A2AError.data is serialized unaltered and exposed publicly to the client in the REST API response.
+            metadata = getattr(error, 'data', None) or {}
+
             return JSONResponse(
                 content={
-                    'message': getattr(error, 'message', str(error)),
-                    'type': type(error).__name__,
+                    'error': {
+                        'code': http_code,
+                        'status': grpc_status,
+                        'message': getattr(error, 'message', str(error)),
+                        'details': [
+                            {
+                                '@type': 'type.googleapis.com/google.rpc.ErrorInfo',
+                                'reason': reason,
+                                'domain': 'a2a-protocol.org',
+                                'metadata': metadata,
+                            }
+                        ],
+                    }
                 },
                 status_code=http_code,
+                media_type='application/json',
             )
         except ParseError as error:
             logger.warning('Parse error: %s', str(error))
             return JSONResponse(
                 content={
-                    'message': str(error),
-                    'type': 'ParseError',
+                    'error': {
+                        'code': 400,
+                        'status': 'INVALID_ARGUMENT',
+                        'message': str(error),
+                        'details': [
+                            {
+                                '@type': 'type.googleapis.com/google.rpc.ErrorInfo',
+                                'reason': 'INVALID_REQUEST',
+                                'domain': 'a2a-protocol.org',
+                                'metadata': {},
+                            }
+                        ],
+                    }
                 },
                 status_code=400,
+                media_type='application/json',
             )
         except Exception:
             logger.exception('Unknown error occurred')
             return JSONResponse(
-                content={'message': 'unknown exception', 'type': 'Exception'},
+                content={
+                    'error': {
+                        'code': 500,
+                        'status': 'INTERNAL',
+                        'message': 'unknown exception',
+                    }
+                },
                 status_code=500,
+                media_type='application/json',
             )
 
     return wrapper

src/a2a/utils/errors.py

@@ -9,6 +9,7 @@ class A2AError(Exception):
     """Base exception for A2A errors."""
 
     message: str = 'A2A Error'
+    data: dict | None = None
 
     def __init__(self, message: str | None = None):
         if message:
@@ -100,6 +101,7 @@ class VersionNotSupportedError(A2AError):
 __all__ = [
     'A2A_ERROR_REASONS',
     'A2A_REASON_TO_ERROR',
+    'A2A_REST_ERROR_MAPPING',
     'JSON_RPC_ERROR_CODE_MAP',
     'ExtensionSupportRequiredError',
     'InternalError',
@@ -132,16 +134,43 @@ class VersionNotSupportedError(A2AError):
 }
 
 
+A2A_REST_ERROR_MAPPING: dict[type[A2AError], tuple[int, str, str]] = {
+    TaskNotFoundError: (404, 'NOT_FOUND', 'TASK_NOT_FOUND'),
+    TaskNotCancelableError: (409, 'FAILED_PRECONDITION', 'TASK_NOT_CANCELABLE'),
+    PushNotificationNotSupportedError: (
+        400,
+        'UNIMPLEMENTED',
+        'PUSH_NOTIFICATION_NOT_SUPPORTED',
+    ),
+    UnsupportedOperationError: (400, 'UNIMPLEMENTED', 'UNSUPPORTED_OPERATION'),
+    ContentTypeNotSupportedError: (
+        415,
+        'INVALID_ARGUMENT',
+        'CONTENT_TYPE_NOT_SUPPORTED',
+    ),
+    InvalidAgentResponseError: (502, 'INTERNAL', 'INVALID_AGENT_RESPONSE'),
+    ExtendedAgentCardNotConfiguredError: (
+        400,
+        'FAILED_PRECONDITION',
+        'EXTENDED_AGENT_CARD_NOT_CONFIGURED',
+    ),
+    ExtensionSupportRequiredError: (
+        400,
+        'FAILED_PRECONDITION',
+        'EXTENSION_SUPPORT_REQUIRED',
+    ),
+    VersionNotSupportedError: (400, 'UNIMPLEMENTED', 'VERSION_NOT_SUPPORTED'),
+    InvalidParamsError: (400, 'INVALID_ARGUMENT', 'INVALID_PARAMS'),
+    InvalidRequestError: (400, 'INVALID_ARGUMENT', 'INVALID_REQUEST'),
+    MethodNotFoundError: (404, 'NOT_FOUND', 'METHOD_NOT_FOUND'),
+    InternalError: (500, 'INTERNAL', 'INTERNAL_ERROR'),
+}
+
+
 A2A_ERROR_REASONS = {
-    TaskNotFoundError: 'TASK_NOT_FOUND',
-    TaskNotCancelableError: 'TASK_NOT_CANCELABLE',
-    PushNotificationNotSupportedError: 'PUSH_NOTIFICATION_NOT_SUPPORTED',
-    UnsupportedOperationError: 'UNSUPPORTED_OPERATION',
-    ContentTypeNotSupportedError: 'CONTENT_TYPE_NOT_SUPPORTED',
-    InvalidAgentResponseError: 'INVALID_AGENT_RESPONSE',
-    ExtendedAgentCardNotConfiguredError: 'EXTENDED_AGENT_CARD_NOT_CONFIGURED',
-    ExtensionSupportRequiredError: 'EXTENSION_SUPPORT_REQUIRED',
-    VersionNotSupportedError: 'VERSION_NOT_SUPPORTED',
+    cls: reason for cls, (_, _, reason) in A2A_REST_ERROR_MAPPING.items()
 }
 
-A2A_REASON_TO_ERROR = {reason: cls for cls, reason in A2A_ERROR_REASONS.items()}
+A2A_REASON_TO_ERROR = {
+    reason: cls for cls, (_, _, reason) in A2A_REST_ERROR_MAPPING.items()
+}