refactor(client): map error responses to domain errors

Re #737

Author: ishymko

src/a2a/client/__init__.py

@@ -13,9 +13,7 @@
 from a2a.client.client_factory import ClientFactory, minimal_agent_card
 from a2a.client.errors import (
     A2AClientError,
-    A2AClientHTTPError,
-    A2AClientJSONError,
-    A2AClientTimeoutError,
+    AgentCardResolutionError,
 )
 from a2a.client.helpers import create_text_message_object
 from a2a.client.middleware import ClientCallContext, ClientCallInterceptor
@@ -27,9 +25,7 @@
 __all__ = [
     'A2ACardResolver',
     'A2AClientError',
-    'A2AClientHTTPError',
-    'A2AClientJSONError',
-    'A2AClientTimeoutError',
+    'AgentCardResolutionError',
     'AuthInterceptor',
     'BaseClient',
     'Client',

src/a2a/client/card_resolver.py

@@ -8,10 +8,7 @@
 
 from google.protobuf.json_format import ParseDict, ParseError
 
-from a2a.client.errors import (
-    A2AClientHTTPError,
-    A2AClientJSONError,
-)
+from a2a.client.errors import AgentCardResolutionError
 from a2a.types.a2a_pb2 import (
     AgentCard,
 )
@@ -64,9 +61,9 @@ async def get_agent_card(
             An `AgentCard` object representing the agent's capabilities.
 
         Raises:
-            A2AClientHTTPError: If an HTTP error occurs during the request.
-            A2AClientJSONError: If the response body cannot be decoded as JSON
-                or validated against the AgentCard schema.
+            A2AClientError: If an HTTP error occurs during the request, if the
+                response body cannot be decoded as JSON, or if it cannot be
+                validated against the AgentCard schema.
         """
         if not relative_card_path:
             # Use the default public agent card path configured during initialization
@@ -92,21 +89,19 @@ async def get_agent_card(
             if signature_verifier:
                 signature_verifier(agent_card)
         except httpx.HTTPStatusError as e:
-            raise A2AClientHTTPError(
-                e.response.status_code,
-                f'Failed to fetch agent card from {target_url}: {e}',
+            raise AgentCardResolutionError(
+                f'Failed to fetch agent card from {target_url} (HTTP {e.response.status_code}): {e}',
             ) from e
         except json.JSONDecodeError as e:
-            raise A2AClientJSONError(
+            raise AgentCardResolutionError(
                 f'Failed to parse JSON for agent card from {target_url}: {e}'
             ) from e
         except httpx.RequestError as e:
-            raise A2AClientHTTPError(
-                503,
-                f'Network communication error fetching agent card from {target_url}: {e}',
+            raise AgentCardResolutionError(
+                f'Network communication error fetching agent card from {target_url} (HTTP 503): {e}',
             ) from e
         except ParseError as e:
-            raise A2AClientJSONError(
+            raise AgentCardResolutionError(
                 f'Failed to validate agent card structure from {target_url}: {e}'
             ) from e
 

src/a2a/client/client_task_manager.py

@@ -1,9 +1,6 @@
 import logging
 
-from a2a.client.errors import (
-    A2AClientInvalidArgsError,
-    A2AClientInvalidStateError,
-)
+from a2a.client.errors import A2AClientError
 from a2a.types.a2a_pb2 import (
     Message,
     StreamResponse,
@@ -53,15 +50,15 @@ def get_task_or_raise(self) -> Task:
             The `Task` object.
 
         Raises:
-            A2AClientInvalidStateError: If there is no current known Task.
+            A2AClientError: If there is no current known Task.
         """
         if not (task := self.get_task()):
             # Note: The source of this error is either from bad client usage
             # or from the server sending invalid updates. It indicates that this
             # task manager has not consumed any information about a task, yet
             # the caller is attempting to retrieve the current state of the task
             # it expects to be present.
-            raise A2AClientInvalidStateError('no current Task')
+            raise A2AClientError('no current Task')
         return task
 
     async def process(
@@ -79,7 +76,7 @@ async def process(
             The updated `Task` object after processing the event.
 
         Raises:
-            ClientError: If the task ID in the event conflicts with the TaskManager's ID
+            A2AClientError: If the task ID in the event conflicts with the TaskManager's ID
                          when the TaskManager's ID is already set.
         """
         if event.HasField('message'):
@@ -88,7 +85,7 @@ async def process(
 
         if event.HasField('task'):
             if self._current_task:
-                raise A2AClientInvalidArgsError(
+                raise A2AClientError(
                     'Task is already set, create new manager for new tasks.'
                 )
             await self._save_task(event.task)

src/a2a/client/errors.py

@@ -1,113 +1,11 @@
 """Custom exceptions for the A2A client."""
 
-from typing import Any
+from a2a.utils.errors import A2AError
 
 
-class A2AClientError(Exception):
+class A2AClientError(A2AError):
     """Base exception for A2A Client errors."""
 
 
-class A2AClientHTTPError(A2AClientError):
-    """Client exception for HTTP errors received from the server."""
-
-    def __init__(self, status_code: int, message: str):
-        """Initializes the A2AClientHTTPError.
-
-        Args:
-            status_code: The HTTP status code of the response.
-            message: A descriptive error message.
-        """
-        self.status_code = status_code
-        self.message = message
-        super().__init__(f'HTTP Error {status_code}: {message}')
-
-    def __repr__(self) -> str:
-        """Returns an unambiguous representation showing structured attributes."""
-        return (
-            f'{self.__class__.__name__}('
-            f'status_code={self.status_code!r}, '
-            f'message={self.message!r})'
-        )
-
-
-class A2AClientJSONError(A2AClientError):
-    """Client exception for JSON errors during response parsing or validation."""
-
-    def __init__(self, message: str):
-        """Initializes the A2AClientJSONError.
-
-        Args:
-            message: A descriptive error message.
-        """
-        self.message = message
-        super().__init__(f'JSON Error: {message}')
-
-    def __repr__(self) -> str:
-        """Returns an unambiguous representation showing structured attributes."""
-        return f'{self.__class__.__name__}(message={self.message!r})'
-
-
-class A2AClientTimeoutError(A2AClientError):
-    """Client exception for timeout errors during a request."""
-
-    def __init__(self, message: str):
-        """Initializes the A2AClientTimeoutError.
-
-        Args:
-            message: A descriptive error message.
-        """
-        self.message = message
-        super().__init__(f'Timeout Error: {message}')
-
-    def __repr__(self) -> str:
-        """Returns an unambiguous representation showing structured attributes."""
-        return f'{self.__class__.__name__}(message={self.message!r})'
-
-
-class A2AClientInvalidArgsError(A2AClientError):
-    """Client exception for invalid arguments passed to a method."""
-
-    def __init__(self, message: str):
-        """Initializes the A2AClientInvalidArgsError.
-
-        Args:
-            message: A descriptive error message.
-        """
-        self.message = message
-        super().__init__(f'Invalid arguments error: {message}')
-
-    def __repr__(self) -> str:
-        """Returns an unambiguous representation showing structured attributes."""
-        return f'{self.__class__.__name__}(message={self.message!r})'
-
-
-class A2AClientInvalidStateError(A2AClientError):
-    """Client exception for an invalid client state."""
-
-    def __init__(self, message: str):
-        """Initializes the A2AClientInvalidStateError.
-
-        Args:
-            message: A descriptive error message.
-        """
-        self.message = message
-        super().__init__(f'Invalid state error: {message}')
-
-    def __repr__(self) -> str:
-        """Returns an unambiguous representation showing structured attributes."""
-        return f'{self.__class__.__name__}(message={self.message!r})'
-
-
-class A2AClientJSONRPCError(A2AClientError):
-    """Client exception for JSON-RPC errors returned by the server."""
-
-    error: dict[str, Any]
-
-    def __init__(self, error: dict[str, Any]):
-        """Initializes the A2AClientJsonRPCError.
-
-        Args:
-            error: The JSON-RPC error dict from the jsonrpc library.
-        """
-        self.error = error
-        super().__init__(f'JSON-RPC Error {self.error}')
+class AgentCardResolutionError(A2AClientError):
+    """Exception raised when an agent card cannot be resolved."""

src/a2a/client/transports/grpc.py

@@ -1,6 +1,12 @@
 import logging
 
 from collections.abc import AsyncGenerator, Callable
+from functools import wraps
+from typing import Any, NoReturn
+
+import a2a.utils.errors
+
+from a2a.client.errors import A2AClientError
 
 
 try:
@@ -43,6 +49,39 @@
 logger = logging.getLogger(__name__)
 
 
+def _map_grpc_error(e: grpc.aio.AioRpcError) -> NoReturn:
+    details = e.details()
+    if isinstance(details, str) and ': ' in details:
+        error_type_name, error_message = details.split(': ', 1)
+        exception_cls = getattr(a2a.utils.errors, error_type_name, None)
+        if (
+            exception_cls
+            and isinstance(exception_cls, type)
+            and issubclass(exception_cls, a2a.utils.errors.A2AError)
+        ):
+            raise exception_cls(error_message) from e
+    raise A2AClientError(f'gRPC Error {e.code().name}: {e.details()}') from e
+
+def _handle_grpc_exception(func: Callable[..., Any]) -> Callable[..., Any]:
+    @wraps(func)
+    async def wrapper(*args: Any, **kwargs: Any) -> Any:
+        try:
+            return await func(*args, **kwargs)
+        except grpc.aio.AioRpcError as e:
+            _map_grpc_error(e)
+    return wrapper
+
+def _handle_grpc_stream_exception(func: Callable[..., Any]) -> Callable[..., Any]:
+    @wraps(func)
+    async def wrapper(*args: Any, **kwargs: Any) -> Any:
+        try:
+            async for item in func(*args, **kwargs):
+                yield item
+        except grpc.aio.AioRpcError as e:
+            _map_grpc_error(e)
+    return wrapper
+
+
 @trace_class(kind=SpanKind.CLIENT)
 class GrpcTransport(ClientTransport):
     """A gRPC transport for the A2A client."""
@@ -87,6 +126,7 @@ def create(
             raise ValueError('grpc_channel_factory is required when using gRPC')
         return cls(config.grpc_channel_factory(url), card, config.extensions)
 
+    @_handle_grpc_exception
     async def send_message(
         self,
         request: SendMessageRequest,
@@ -100,6 +140,7 @@ async def send_message(
             metadata=self._get_grpc_metadata(extensions),
         )
 
+    @_handle_grpc_stream_exception
     async def send_message_streaming(
         self,
         request: SendMessageRequest,
@@ -118,6 +159,7 @@ async def send_message_streaming(
                 break
             yield response
 
+    @_handle_grpc_stream_exception
     async def subscribe(
         self,
         request: SubscribeToTaskRequest,
@@ -136,6 +178,7 @@ async def subscribe(
                 break
             yield response
 
+    @_handle_grpc_exception
     async def get_task(
         self,
         request: GetTaskRequest,
@@ -149,6 +192,7 @@ async def get_task(
             metadata=self._get_grpc_metadata(extensions),
         )
 
+    @_handle_grpc_exception
     async def list_tasks(
         self,
         request: ListTasksRequest,
@@ -162,6 +206,7 @@ async def list_tasks(
             metadata=self._get_grpc_metadata(extensions),
         )
 
+    @_handle_grpc_exception
     async def cancel_task(
         self,
         request: CancelTaskRequest,
@@ -175,6 +220,7 @@ async def cancel_task(
             metadata=self._get_grpc_metadata(extensions),
         )
 
+    @_handle_grpc_exception
     async def create_task_push_notification_config(
         self,
         request: CreateTaskPushNotificationConfigRequest,
@@ -188,6 +234,7 @@ async def create_task_push_notification_config(
             metadata=self._get_grpc_metadata(extensions),
         )
 
+    @_handle_grpc_exception
     async def get_task_push_notification_config(
         self,
         request: GetTaskPushNotificationConfigRequest,
@@ -201,6 +248,7 @@ async def get_task_push_notification_config(
             metadata=self._get_grpc_metadata(extensions),
         )
 
+    @_handle_grpc_exception
     async def list_task_push_notification_configs(
         self,
         request: ListTaskPushNotificationConfigsRequest,
@@ -214,6 +262,7 @@ async def list_task_push_notification_configs(
             metadata=self._get_grpc_metadata(extensions),
         )
 
+    @_handle_grpc_exception
     async def delete_task_push_notification_config(
         self,
         request: DeleteTaskPushNotificationConfigRequest,
@@ -227,6 +276,7 @@ async def delete_task_push_notification_config(
             metadata=self._get_grpc_metadata(extensions),
         )
 
+    @_handle_grpc_exception
     async def get_extended_agent_card(
         self,
         *,