diff --git a/pyproject.toml b/pyproject.toml index dc64177b1..c1fbb9129 100644 --- a/pyproject.toml +++ b/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 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), diff --git a/src/a2a/server/request_handlers/grpc_handler.py b/src/a2a/server/request_handlers/grpc_handler.py index fc7e03084..d4d8ed669 100644 --- a/src/a2a/server/request_handlers/grpc_handler.py +++ b/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() diff --git a/src/a2a/utils/grpc_status.py b/src/a2a/utils/grpc_status.py new file mode 100644 index 000000000..5f0074f24 --- /dev/null +++ b/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 diff --git a/tests/utils/test_grpc_status.py b/tests/utils/test_grpc_status.py new file mode 100644 index 000000000..e62267963 --- /dev/null +++ b/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) diff --git a/uv.lock b/uv.lock index 973555fc4..d09af8654 100644 --- a/uv.lock +++ b/uv.lock @@ -30,7 +30,6 @@ all = [ { name = "fastapi" }, { name = "grpcio" }, { name = "grpcio-reflection" }, - { name = "grpcio-status" }, { name = "grpcio-tools" }, { name = "opentelemetry-api" }, { name = "opentelemetry-sdk" }, @@ -53,7 +52,6 @@ fastapi = [ grpc = [ { name = "grpcio" }, { name = "grpcio-reflection" }, - { name = "grpcio-status" }, { name = "grpcio-tools" }, ] http-server = [ @@ -117,8 +115,6 @@ requires-dist = [ { name = "grpcio", marker = "extra == 'grpc'", specifier = ">=1.60" }, { name = "grpcio-reflection", marker = "extra == 'all'", specifier = ">=1.7.0" }, { name = "grpcio-reflection", marker = "extra == 'grpc'", specifier = ">=1.7.0" }, - { name = "grpcio-status", marker = "extra == 'all'", specifier = ">=1.60" }, - { name = "grpcio-status", marker = "extra == 'grpc'", specifier = ">=1.60" }, { name = "grpcio-tools", marker = "extra == 'all'", specifier = ">=1.60" }, { name = "grpcio-tools", marker = "extra == 'grpc'", specifier = ">=1.60" }, { name = "httpx", specifier = ">=0.28.1" }, @@ -1000,20 +996,6 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/d5/61/e472357ff5484f67c802568e70b3182df3d8eb1d0c2de38199d9a9a28bb2/grpcio_reflection-1.81.0-py3-none-any.whl", hash = "sha256:85322a9c1ab62d9823b1262a9d78d653b1710b99b5764cdcef2673cfe352b9c1", size = 22907, upload-time = "2026-06-01T06:00:16.714Z" }, ] -[[package]] -name = "grpcio-status" -version = "1.81.0" -source = { registry = "https://pypi.org/simple" } -dependencies = [ - { name = "googleapis-common-protos" }, - { name = "grpcio" }, - { name = "protobuf" }, -] -sdist = { url = "https://files.pythonhosted.org/packages/01/b6/cdc177114997d15c887fb09ccfd16705c8ceb8b4ca2487902b54a7bfd1af/grpcio_status-1.81.0.tar.gz", hash = "sha256:b6fe9788cfdd1f0f63c0528a1e0bfdb41e8ff0583e920d2d8e8888598c01bb69", size = 13900, upload-time = "2026-06-01T06:00:32.638Z" } -wheels = [ - { url = "https://files.pythonhosted.org/packages/f3/b7/5aa346bf1cdecd4ed64b86c10a4d5a089ce3da89145f8328caf0b22b240d/grpcio_status-1.81.0-py3-none-any.whl", hash = "sha256:10eb4c2309db902dc26c1873e80a821bf794be772c10dfd83030f7f59f165fab", size = 14634, upload-time = "2026-06-01T06:00:13.345Z" }, -] - [[package]] name = "grpcio-tools" version = "1.81.0"