refactor(grpc): remove grpcio-status dependency

pyproject.toml

@@ -37,7 +37,7 @@ classifiers = [
 http-server = ["sse-starlette", "starlette"]
 fastapi = ["a2a-sdk[http-server]", "fastapi>=0.115.2"]
 encryption = ["cryptography>=43.0.0"]
-grpc = ["grpcio>=1.60", "grpcio-tools>=1.60", "grpcio-status>=1.60", "grpcio_reflection>=1.7.0"]
+grpc = ["grpcio>=1.60", "grpcio-tools>=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"]
@@ -94,7 +94,7 @@ filterwarnings = [
   # ResourceWarnings from asyncio event loop/socket cleanup during garbage collection
   # These appear intermittently between tests due to pytest-asyncio and sse-starlette timing
   "ignore:unclosed event loop:ResourceWarning",
-  "ignore:unclosed transport:ResourceWarning", 
+  "ignore:unclosed transport:ResourceWarning",
   "ignore:unclosed <socket.socket:ResourceWarning",
 ]
 

src/a2a/client/transports/grpc.py

@@ -10,11 +10,9 @@
 
 try:
     import grpc  # type: ignore[reportMissingModuleSource]
-
-    from grpc_status import rpc_status
 except ImportError as e:
     raise ImportError(
-        'A2AGrpcClient requires grpcio, grpcio-tools, and grpcio-status to be installed. '
+        'A2AGrpcClient requires grpcio and grpcio-tools to be installed. '
         'Install with: '
         "'pip install a2a-sdk[grpc]'"
     ) from e
@@ -48,6 +46,7 @@
 )
 from a2a.utils.constants import PROTOCOL_VERSION_CURRENT, VERSION_HEADER
 from a2a.utils.errors import A2A_REASON_TO_ERROR, A2AError
+from a2a.utils.grpc_status import status_from_call
 from a2a.utils.proto_utils import bad_request_to_validation_errors
 from a2a.utils.telemetry import SpanKind, trace_class
 
@@ -56,12 +55,10 @@
 
 
 def _map_grpc_error(e: grpc.aio.AioRpcError) -> NoReturn:
-
     if e.code() == grpc.StatusCode.DEADLINE_EXCEEDED:
         raise A2AClientTimeoutError('Client Request timed out') from e
 
-    # Use grpc_status to cleanly extract the rich Status from the call
-    status = rpc_status.from_call(cast('grpc.Call', e))
+    status = status_from_call(cast('grpc.Call', e))
     data = None
 
     if status is not None:
@@ -321,7 +318,6 @@ async def _call_grpc(
         context: ClientCallContext | None,
         **kwargs: Any,
     ) -> Any:
-
         return await method(
             request,
             metadata=self._get_grpc_metadata(context),
@@ -336,7 +332,6 @@ async def _call_grpc_stream(
         context: ClientCallContext | None,
         **kwargs: Any,
     ) -> AsyncGenerator[StreamResponse]:
-
         stream = method(
             request,
             metadata=self._get_grpc_metadata(context),

src/a2a/server/request_handlers/grpc_handler.py

@@ -9,11 +9,9 @@
 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, grpcio-tools, and grpcio-status to be installed. '
+        'GrpcHandler requires grpcio and grpcio-tools to be installed. '
         'Install with: '
         "'pip install a2a-sdk[grpc]'"
     ) from e
@@ -34,6 +32,7 @@
 from a2a.types import a2a_pb2
 from a2a.utils import proto_utils
 from a2a.utils.errors import A2A_ERROR_REASONS, A2AError, TaskNotFoundError
+from a2a.utils.grpc_status import status_to_grpc
 from a2a.utils.proto_utils import validation_errors_to_bad_request
 
 
@@ -400,8 +399,7 @@ async def abort_context(
                 )
                 status.details.append(bad_request_detail)
 
-            # Use grpc_status to safely generate standard trailing metadata
-            rich_status = rpc_status.to_status(status)
+            rich_status = status_to_grpc(status)
 
             new_metadata: list[tuple[str, str | bytes]] = []
             trailing = context.trailing_metadata()

src/a2a/utils/grpc_status.py

@@ -0,0 +1,71 @@
+import logging
+
+from typing import Any, NamedTuple
+
+
+try:
+    import grpc  # type: ignore[reportMissingModuleSource]
+except ImportError:
+    grpc = None  # type: ignore
+
+from google.rpc import status_pb2  # type: ignore[reportMissingModuleSource]
+
+
+logger = logging.getLogger(__name__)
+
+GRPC_STATUS_DETAILS_BIN_KEY = 'grpc-status-details-bin'
+
+
+class GrpcStatus(NamedTuple):
+    """Represents the gRPC status code, details, and trailing metadata."""
+
+    code: Any
+    details: str
+    trailing_metadata: tuple
+
+
+def status_to_grpc(status: status_pb2.Status) -> GrpcStatus:
+    """Converts a google.rpc.status.Status message into its gRPC components."""
+    if grpc is None:
+        raise ImportError(
+            'gRPC is not installed. Install with: pip install a2a-sdk[grpc]'
+        )
+    for x in grpc.StatusCode:
+        if x.value[0] == status.code:
+            grpc_code = x
+            break
+    else:
+        raise ValueError(f'Invalid status code {status.code}')
+
+    bin_data = status.SerializeToString()
+    metadata = ((GRPC_STATUS_DETAILS_BIN_KEY, bin_data),)
+
+    return GrpcStatus(grpc_code, status.message, metadata)
+
+
+def status_from_call(call: Any) -> status_pb2.Status | None:
+    """Extracts a google.rpc.status.Status message from a grpc.Call instance."""
+    if grpc is None:
+        raise ImportError(
+            'gRPC is not installed. Install with: pip install a2a-sdk[grpc]'
+        )
+    if not hasattr(call, 'trailing_metadata'):
+        return None
+    trailing_metadata = call.trailing_metadata()
+    if not trailing_metadata:
+        return None
+    for k, v in trailing_metadata:
+        if k == GRPC_STATUS_DETAILS_BIN_KEY:
+            status = status_pb2.Status()
+            status.ParseFromString(v)
+
+            if call.code().value[0] != status.code:
+                raise ValueError(
+                    f'Mismatched status code: proto={status.code}, call={call.code()}'
+                )
+            if call.details() != status.message:
+                raise ValueError(
+                    f'Mismatched status message: proto={status.message!r}, call={call.details()!r}'
+                )
+            return status
+    return None

tests/utils/test_grpc_status.py

@@ -0,0 +1,143 @@
+import grpc
+import pytest
+
+from a2a.utils.grpc_status import (
+    GRPC_STATUS_DETAILS_BIN_KEY,
+    status_from_call,
+    status_to_grpc,
+)
+from google.protobuf import any_pb2
+from google.rpc import error_details_pb2, status_pb2
+
+
+class MockGrpcCall:
+    """Mock for a gRPC Call object simulating trailing metadata and status info."""
+
+    def __init__(
+        self, code: grpc.StatusCode, details: str, trailing_metadata: tuple
+    ):
+        self._code = code
+        self._details = details
+        self._trailing_metadata = trailing_metadata
+
+    def code(self) -> grpc.StatusCode:
+        return self._code
+
+    def details(self) -> str:
+        return self._details
+
+    def trailing_metadata(self) -> tuple:
+        return self._trailing_metadata
+
+
+def test_status_to_grpc_success():
+    """Test that a Status protobuf is correctly mapped to GrpcStatus namedtuple."""
+    status = status_pb2.Status(
+        code=grpc.StatusCode.INVALID_ARGUMENT.value[0],
+        message='Invalid parameter provided',
+    )
+
+    grpc_status = status_to_grpc(status)
+
+    assert grpc_status.code == grpc.StatusCode.INVALID_ARGUMENT
+    assert grpc_status.details == 'Invalid parameter provided'
+    assert len(grpc_status.trailing_metadata) == 1
+
+    key, val = grpc_status.trailing_metadata[0]
+    assert key == GRPC_STATUS_DETAILS_BIN_KEY
+
+    # Parse back the serialized bytes to verify content
+    parsed_status = status_pb2.Status()
+    parsed_status.ParseFromString(val)
+    assert parsed_status.code == status.code
+    assert parsed_status.message == status.message
+
+
+def test_status_to_grpc_invalid_code():
+    """Test that ValueError is raised for an invalid gRPC status code value."""
+    status = status_pb2.Status(code=999, message='Bad status code')
+    with pytest.raises(ValueError, match='Invalid status code 999'):
+        status_to_grpc(status)
+
+
+def test_status_from_call_no_metadata():
+    """Test that status_from_call returns None if trailing metadata is missing."""
+    call = MockGrpcCall(grpc.StatusCode.OK, 'OK', ())
+    assert status_from_call(call) is None
+
+
+def test_status_from_call_success_roundtrip():
+    """Test standard roundtrip: Status -> status_to_grpc -> Call -> status_from_call."""
+    # 1. Create a status with error details
+    status = status_pb2.Status(
+        code=grpc.StatusCode.NOT_FOUND.value[0],
+        message='Task not found',
+    )
+    error_info = error_details_pb2.ErrorInfo(
+        reason='TASK_NOT_FOUND',
+        domain='a2a-protocol.org',
+    )
+    detail = any_pb2.Any()
+    detail.Pack(error_info)
+    status.details.append(detail)
+
+    # 2. Convert to gRPC components
+    grpc_status = status_to_grpc(status)
+
+    # 3. Wrap in a mock Call object
+    call = MockGrpcCall(
+        code=grpc_status.code,
+        details=grpc_status.details,
+        trailing_metadata=grpc_status.trailing_metadata,
+    )
+
+    # 4. Parse back using status_from_call
+    parsed_status = status_from_call(call)
+
+    assert parsed_status is not None
+    assert parsed_status.code == status.code
+    assert parsed_status.message == status.message
+    assert len(parsed_status.details) == 1
+
+    parsed_detail = error_details_pb2.ErrorInfo()
+    parsed_status.details[0].Unpack(parsed_detail)
+    assert parsed_detail.reason == 'TASK_NOT_FOUND'
+    assert parsed_detail.domain == 'a2a-protocol.org'
+
+
+def test_status_from_call_mismatched_code():
+    """Test that ValueError is raised if call status code doesn't match parsed status code."""
+    status = status_pb2.Status(
+        code=grpc.StatusCode.INVALID_ARGUMENT.value[0],
+        message='Mismatched status code test',
+    )
+    grpc_status = status_to_grpc(status)
+
+    # Intentionally change the mock Call code to StatusCode.UNAUTHENTICATED
+    call = MockGrpcCall(
+        code=grpc.StatusCode.UNAUTHENTICATED,
+        details=grpc_status.details,
+        trailing_metadata=grpc_status.trailing_metadata,
+    )
+
+    with pytest.raises(ValueError, match='Mismatched status code'):
+        status_from_call(call)
+
+
+def test_status_from_call_mismatched_message():
+    """Test that ValueError is raised if call details message doesn't match parsed status message."""
+    status = status_pb2.Status(
+        code=grpc.StatusCode.INVALID_ARGUMENT.value[0],
+        message='Mismatched status message test',
+    )
+    grpc_status = status_to_grpc(status)
+
+    # Intentionally change the mock Call details message
+    call = MockGrpcCall(
+        code=grpc_status.code,
+        details='A completely different message',
+        trailing_metadata=grpc_status.trailing_metadata,
+    )
+
+    with pytest.raises(ValueError, match='Mismatched status message'):
+        status_from_call(call)