feat(rest): update REST error handling to use google.rpc.Status (#838)

# Description
This PR refactors the REST transport error handling to adhere to the
`google.rpc.Status` JSON format. Both the server-side exception handlers
and the client-side REST transport have been updated to utilize the new
standard error envelope, ensuring consistency across A2A REST APIs.

## Summary of Changes

* **Server:** 
* Updated `rest_error_handler` and the global `StarletteHTTPException`
handler in `A2ARESTFastAPIApplication` to return errors wrapped in an
`{'error': {...}}` envelope.
* Payloads now correctly include the HTTP `code`, gRPC `status`,
`message`, and a `details` array containing
`type.googleapis.com/google.rpc.ErrorInfo` for the specific reason and
metadata.
* **Client:** 
* Modified `RestTransport._handle_http_error` to parse the new format.
It now gracefully extracts the `reason` from the `ErrorInfo` detail
object to map it back to the corresponding Python `A2AError` class.
* **Core/Utils:** 
* Introduced `A2A_REST_ERROR_MAPPING` in `errors.py` to centralize the
mapping of Python exceptions to their respective HTTP status codes, gRPC
statuses, and string reasons.
* Added a `data` attribute to the base `A2AError` to carry arbitrary
error metadata.
* **Tests:** 
* Updated REST client, server, and error handler tests to validate the
new nested `{'error': {...}}` JSON payload structures.


- [X] Follow the [`CONTRIBUTING`
Guide](https://github.com/a2aproject/a2a-python/blob/main/CONTRIBUTING.md).
- [X] Make your Pull Request title in the
<https://www.conventionalcommits.org/> specification.
- Important Prefixes for
[release-please](https://github.com/googleapis/release-please):
- `fix:` which represents bug fixes, and correlates to a
[SemVer](https://semver.org/) patch.
- `feat:` represents a new feature, and correlates to a SemVer minor.
- `feat!:`, or `fix!:`, `refactor!:`, etc., which represent a breaking
change (indicated by the `!`) and will result in a SemVer major.
- [X] Ensure the tests and linter pass (Run `bash scripts/format.sh`
from the repository root to format)
- [X] Appropriate docs were updated (if necessary)

Fixes #722  🦕

---------

Co-authored-by: Ivan Shymko <ishymko@google.com>

Author: knapg

.github/actions/spelling/allow.txt

@@ -14,6 +14,7 @@ agentic
 AGrpc
 aio
 aiomysql
+AIP
 alg
 amannn
 aproject

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,70 +17,40 @@
 
 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,
+    RestErrorMap,
 )
 
 
 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 _build_error_payload(
+    code: int,
+    status: str,
+    message: str,
+    reason: str | None = None,
+    metadata: dict[str, Any] | None = None,
+) -> dict[str, Any]:
+    """Helper function to build the JSON error payload."""
+    payload: dict[str, Any] = {
+        'code': code,
+        'status': status,
+        'message': message,
+    }
+    if reason:
+        payload['details'] = [
+            {
+                '@type': 'type.googleapis.com/google.rpc.ErrorInfo',
+                'reason': reason,
+                'domain': 'a2a-protocol.org',
+                'metadata': metadata if metadata is not None else {},
+            }
+        ]
+    return {'error': payload}
 
 
 def rest_error_handler(
@@ -93,9 +63,12 @@ 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
+            mapping = A2A_REST_ERROR_MAPPING.get(
+                type(error), RestErrorMap(500, 'INTERNAL', 'INTERNAL_ERROR')
             )
+            http_code = mapping.http_code
+            grpc_status = mapping.grpc_status
+            reason = mapping.reason
 
             log_level = (
                 logging.ERROR
@@ -107,32 +80,46 @@ 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 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__,
-                },
+                content=_build_error_payload(
+                    code=http_code,
+                    status=grpc_status,
+                    message=getattr(error, 'message', str(error)),
+                    reason=reason,
+                    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',
-                },
+                content=_build_error_payload(
+                    code=400,
+                    status='INVALID_ARGUMENT',
+                    message=str(error),
+                    reason='INVALID_REQUEST',
+                    metadata={},
+                ),
                 status_code=400,
+                media_type='application/json',
             )
         except Exception:
             logger.exception('Unknown error occurred')
             return JSONResponse(
-                content={'message': 'unknown exception', 'type': 'Exception'},
+                content=_build_error_payload(
+                    code=500,
+                    status='INTERNAL',
+                    message='unknown exception',
+                ),
                 status_code=500,
+                media_type='application/json',
             )
 
     return wrapper
@@ -158,9 +145,7 @@ async def wrapper(*args: Any, **kwargs: Any) -> Any:
                 "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 error.data else '',
             )
             # Since the stream has started, we can't return a JSONResponse.
             # Instead, we run the error handling logic (provides logging)