a2aproject · Iwaniukooo11 · Jul 10, 2026 · Jul 10, 2026
@@ -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",
 ]
 

@@ -10,18 +10,17 @@
 
 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
 
 
 from google.rpc import (  # type: ignore[reportMissingModuleSource]
     error_details_pb2,
+    status_pb2,
 )
 
 from a2a.client.client import ClientConfig
@@ -55,13 +54,35 @@
 logger = logging.getLogger(__name__)
 
 
-def _map_grpc_error(e: grpc.aio.AioRpcError) -> NoReturn:
+def _from_call(call: grpc.Call) -> status_pb2.Status | None:
+    """Extracts a google.rpc.status.Status message from a grpc.Call instance."""
+    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':
+            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
+
 
+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 = _from_call(cast('grpc.Call', e))
     data = None
 
     if status is not None:
@@ -321,7 +342,6 @@ async def _call_grpc(
         context: ClientCallContext | None,
         **kwargs: Any,
     ) -> Any:
-
         return await method(
             request,
             metadata=self._get_grpc_metadata(context),
@@ -336,7 +356,6 @@ async def _call_grpc_stream(
         context: ClientCallContext | None,
         **kwargs: Any,
     ) -> AsyncGenerator[StreamResponse]:
-
         stream = method(
             request,
             metadata=self._get_grpc_metadata(context),

@@ -3,17 +3,15 @@
 
 from abc import ABC, abstractmethod
 from collections.abc import AsyncIterable, Awaitable, Callable
-from typing import TypeVar
+from typing import NamedTuple, TypeVar
 
 
 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
@@ -37,6 +35,29 @@
 from a2a.utils.proto_utils import validation_errors_to_bad_request
 
 
+class GrpcStatus(NamedTuple):
+    """Represents the gRPC status code, details, and trailing metadata."""
+
+    code: grpc.StatusCode
+    details: str
+    trailing_metadata: tuple
+
+
+def _to_status(status: status_pb2.Status) -> GrpcStatus:
+    """Converts a google.rpc.status.Status message into its gRPC components."""
+    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', bin_data),)
+
+    return GrpcStatus(grpc_code, status.message, metadata)
+
+
 logger = logging.getLogger(__name__)
 
 
@@ -400,8 +421,8 @@ 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)
+            # Use local helper to safely generate standard trailing metadata
+            rich_status = _to_status(status)
 
             new_metadata: list[tuple[str, str | bytes]] = []
             trailing = context.trailing_metadata()