fix: clean up legacy grpc error mapping and status handling

- Removed unreleased legacy string-split fallback parsing for gRPC errors.
- Refactored client and server handlers to use pure grpcio-status helpers for
  encoding/decoding google.rpc.Status to standard trailing metadata.
- Cleaned up A2A_ERROR_REASONS map to strictly contain only the 9 domain
  errors defined in the specification (removing generic JSON-RPC errors).
- Added grpcio-status to pyproject.toml dependencies.

Author: knapg

.jscpd.json

@@ -1,5 +1,13 @@
 {
-  "ignore": ["**/.github/**", "**/.git/**", "**/tests/**", "**/src/a2a/grpc/**", "**/.nox/**", "**/.venv/**"],
+  "ignore": [
+    "**/.github/**",
+    "**/.git/**",
+    "**/tests/**",
+    "**/src/a2a/grpc/**",
+    "**/src/a2a/compat/**",
+    "**/.nox/**",
+    "**/.venv/**"
+  ],
   "threshold": 3,
   "reporters": ["html", "markdown"]
 }

pyproject.toml

@@ -33,7 +33,7 @@ classifiers = [
 [project.optional-dependencies]
 http-server = ["fastapi>=0.115.2", "sse-starlette", "starlette"]
 encryption = ["cryptography>=43.0.0"]
-grpc = ["grpcio>=1.60", "grpcio-tools>=1.60", "grpcio_reflection>=1.7.0"]
+grpc = ["grpcio>=1.60", "grpcio-tools>=1.60", "grpcio-status>=1.60", "grpcio_reflection>=1.7.0"]
 telemetry = ["opentelemetry-api>=1.33.0", "opentelemetry-sdk>=1.33.0"]
 postgresql = ["sqlalchemy[asyncio,postgresql-asyncpg]>=2.0.0"]
 mysql = ["sqlalchemy[asyncio,aiomysql]>=2.0.0"]

src/a2a/client/transports/grpc.py

@@ -1,27 +1,27 @@
 import logging
 
-from collections.abc import AsyncGenerator, Callable, Iterable
+from collections.abc import AsyncGenerator, Callable
 from functools import wraps
 from typing import Any, NoReturn, cast
 
 from a2a.client.errors import A2AClientError, A2AClientTimeoutError
 from a2a.client.middleware import ClientCallContext
-from a2a.utils.errors import JSON_RPC_ERROR_CODE_MAP, A2AError
 
 
 try:
     import grpc  # type: ignore[reportMissingModuleSource]
+
+    from grpc_status import rpc_status
 except ImportError as e:
     raise ImportError(
-        'A2AGrpcClient requires grpcio and grpcio-tools to be installed. '
+        'A2AGrpcClient requires grpcio, grpcio-tools, and grpcio-status to be installed. '
         'Install with: '
         "'pip install a2a-sdk[grpc]'"
     ) from e
 
 
 from google.rpc import (  # type: ignore[reportMissingModuleSource]
     error_details_pb2,
-    status_pb2,
 )
 
 from a2a.client.client import ClientConfig
@@ -55,16 +55,16 @@
 
 logger = logging.getLogger(__name__)
 
-_A2A_ERROR_NAME_TO_CLS = {
-    error_type.__name__: error_type for error_type in JSON_RPC_ERROR_CODE_MAP
-}
 
+def _map_grpc_error(e: grpc.aio.AioRpcError) -> NoReturn:
+
+    if e.code() == grpc.StatusCode.DEADLINE_EXCEEDED:
+        raise A2AClientTimeoutError('Client Request timed out') from e
+
+    # 1. Use grpc_status to cleanly extract the rich Status from the call
+    status = rpc_status.from_call(cast('grpc.Call', e))
 
-def _parse_rich_grpc_error(
-    value: bytes, original_error: grpc.aio.AioRpcError
-) -> None:
-    try:
-        status = status_pb2.Status.FromString(value)
+    if status is not None:
         for detail in status.details:
             if detail.Is(error_details_pb2.ErrorInfo.DESCRIPTOR):
                 error_info = error_details_pb2.ErrorInfo()
@@ -73,34 +73,8 @@ def _parse_rich_grpc_error(
                 if error_info.domain == 'a2a-protocol.org':
                     exception_cls = A2A_REASON_TO_ERROR.get(error_info.reason)
                     if exception_cls:
-                        raise exception_cls(status.message) from original_error  # noqa: TRY301
-    except Exception as parse_e:
-        # Don't swallow A2A errors generated above
-        if isinstance(parse_e, (A2AError, A2AClientError)):
-            raise parse_e
-        logger.warning(
-            'Failed to parse grpc-status-details-bin', exc_info=parse_e
-        )
-
-
-def _map_grpc_error(e: grpc.aio.AioRpcError) -> NoReturn:
-    if e.code() == grpc.StatusCode.DEADLINE_EXCEEDED:
-        raise A2AClientTimeoutError('Client Request timed out') from e
+                        raise exception_cls(status.message) from e
 
-    metadata = e.trailing_metadata()
-    if metadata:
-        iterable_metadata = cast('Iterable[tuple[str, str | bytes]]', metadata)
-        for key, value in iterable_metadata:
-            if key == 'grpc-status-details-bin' and isinstance(value, bytes):
-                _parse_rich_grpc_error(value, e)
-
-    details = e.details()
-    if isinstance(details, str) and ': ' in details:
-        error_type_name, error_message = details.split(': ', 1)
-        # Leaving as fallback for errors that don't use the rich error details.
-        exception_cls = _A2A_ERROR_NAME_TO_CLS.get(error_type_name)
-        if exception_cls:
-            raise exception_cls(error_message) from e
     raise A2AClientError(f'gRPC Error {e.code().name}: {e.details()}') from e
 
 

src/a2a/server/request_handlers/grpc_handler.py

@@ -4,15 +4,16 @@
 
 from abc import ABC, abstractmethod
 from collections.abc import AsyncIterable, Awaitable, Callable
-from typing import cast
 
 
 try:
     import grpc  # type: ignore[reportMissingModuleSource]
     import grpc.aio  # type: ignore[reportMissingModuleSource]
+
+    from grpc_status import rpc_status
 except ImportError as e:
     raise ImportError(
-        'GrpcHandler requires grpcio and grpcio-tools to be installed. '
+        'GrpcHandler requires grpcio, grpcio-tools, and grpcio-status to be installed. '
         'Install with: '
         "'pip install a2a-sdk[grpc]'"
     ) from e
@@ -420,37 +421,40 @@ async def abort_context(
         """Sets the grpc errors appropriately in the context."""
         code = _ERROR_CODE_MAP.get(type(error))
 
-        status_value = code.value if code else grpc.StatusCode.UNKNOWN.value
-        status_code = (
-            status_value[0] if isinstance(status_value, tuple) else status_value
-        )
-        error_msg = error.message if hasattr(error, 'message') else str(error)
-        status = status_pb2.Status(code=status_code, message=error_msg)
-
         if code:
             reason = A2A_ERROR_REASONS.get(type(error), 'UNKNOWN_ERROR')
-
             error_info = error_details_pb2.ErrorInfo(
                 reason=reason,
                 domain='a2a-protocol.org',
             )
 
+            status_code = (
+                code.value[0] if code else grpc.StatusCode.UNKNOWN.value[0]
+            )
+            error_msg = (
+                error.message if hasattr(error, 'message') else str(error)
+            )
+
+            # Create standard Status and pack the ErrorInfo
+            status = status_pb2.Status(code=status_code, message=error_msg)
             detail = any_pb2.Any()
             detail.Pack(error_info)
             status.details.append(detail)
 
-        context.set_trailing_metadata(
-            cast(
-                'tuple[tuple[str, str | bytes], ...]',
-                (('grpc-status-details-bin', status.SerializeToString()),),
-            )
-        )
+            # Use grpc_status to safely generate standard trailing metadata
+            rich_status = rpc_status.to_status(status)
 
-        if code:
-            await context.abort(
-                code,
-                status.message,
-            )
+            new_metadata: list[tuple[str, str | bytes]] = []
+            trailing = context.trailing_metadata()
+            if trailing:
+                for k, v in trailing:
+                    new_metadata.append((str(k), v))
+
+            for k, v in rich_status.trailing_metadata:
+                new_metadata.append((str(k), v))
+
+            context.set_trailing_metadata(tuple(new_metadata))
+            await context.abort(rich_status.code, rich_status.details)
         else:
             await context.abort(
                 grpc.StatusCode.UNKNOWN,

src/a2a/server/request_handlers/response_helpers.py

@@ -27,6 +27,7 @@
     SendMessageResponse as SendMessageResponseProto,
 )
 from a2a.utils.errors import (
+    JSON_RPC_ERROR_CODE_MAP,
     A2AError,
     AuthenticatedExtendedCardNotConfiguredError,
     ContentTypeNotSupportedError,
@@ -56,19 +57,6 @@
     InternalError: JSONRPCInternalError,
 }
 
-ERROR_CODE_MAP: dict[type[A2AError], int] = {
-    TaskNotFoundError: -32001,
-    TaskNotCancelableError: -32002,
-    PushNotificationNotSupportedError: -32003,
-    UnsupportedOperationError: -32004,
-    ContentTypeNotSupportedError: -32005,
-    InvalidAgentResponseError: -32006,
-    AuthenticatedExtendedCardNotConfiguredError: -32007,
-    InvalidParamsError: -32602,
-    InvalidRequestError: -32600,
-    MethodNotFoundError: -32601,
-}
-
 
 # Tuple of all A2AError types for isinstance checks
 _A2A_ERROR_TYPES: tuple[type, ...] = (A2AError,)
@@ -136,7 +124,7 @@ def build_error_response(
     elif isinstance(error, A2AError):
         error_type = type(error)
         model_class = EXCEPTION_MAP.get(error_type, JSONRPCInternalError)
-        code = ERROR_CODE_MAP.get(error_type, -32603)
+        code = JSON_RPC_ERROR_CODE_MAP.get(error_type, -32603)
         jsonrpc_error = model_class(
             code=code,
             message=str(error),

src/a2a/utils/errors.py

@@ -140,10 +140,6 @@ class VersionNotSupportedError(A2AError):
     AuthenticatedExtendedCardNotConfiguredError: 'EXTENDED_AGENT_CARD_NOT_CONFIGURED',
     ExtensionSupportRequiredError: 'EXTENSION_SUPPORT_REQUIRED',
     VersionNotSupportedError: 'VERSION_NOT_SUPPORTED',
-    InvalidParamsError: 'INVALID_PARAMS',
-    InvalidRequestError: 'INVALID_REQUEST',
-    MethodNotFoundError: 'METHOD_NOT_FOUND',
-    InternalError: 'INTERNAL_ERROR',
 }
 
 A2A_REASON_TO_ERROR = {reason: cls for cls, reason in A2A_ERROR_REASONS.items()}

tests/client/transports/test_grpc_client.py

@@ -36,7 +36,6 @@
     TaskStatusUpdateEvent,
 )
 from a2a.utils import get_text_parts
-from a2a.utils.errors import JSON_RPC_ERROR_CODE_MAP
 
 
 @pytest.fixture
@@ -259,29 +258,7 @@ async def test_send_message_with_timeout_context(
     assert kwargs['timeout'] == 12.5
 
 
-@pytest.mark.parametrize('error_cls', list(JSON_RPC_ERROR_CODE_MAP.keys()))
-@pytest.mark.asyncio
-async def test_grpc_mapped_errors_legacy(
-    grpc_transport: GrpcTransport,
-    mock_grpc_stub: AsyncMock,
-    sample_message_send_params: SendMessageRequest,
-    error_cls,
-) -> None:
-    """Test handling of legacy gRPC error responses."""
-    error_details = f'{error_cls.__name__}: Mapped Error'
-
-    mock_grpc_stub.SendMessage.side_effect = grpc.aio.AioRpcError(
-        code=grpc.StatusCode.INTERNAL,
-        initial_metadata=grpc.aio.Metadata(),
-        trailing_metadata=grpc.aio.Metadata(),
-        details=error_details,
-    )
-
-    with pytest.raises(error_cls):
-        await grpc_transport.send_message(sample_message_send_params)
-
-
-@pytest.mark.parametrize('error_cls', list(JSON_RPC_ERROR_CODE_MAP.keys()))
+@pytest.mark.parametrize('error_cls', list(A2A_ERROR_REASONS.keys()))
 @pytest.mark.asyncio
 async def test_grpc_mapped_errors_rich(
     grpc_transport: GrpcTransport,
@@ -312,7 +289,7 @@ async def test_grpc_mapped_errors_rich(
         trailing_metadata=grpc.aio.Metadata(
             ('grpc-status-details-bin', status.SerializeToString()),
         ),
-        details='A generic error message',
+        details=error_details,
     )
 
     with pytest.raises(error_cls) as excinfo: