Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
69 changes: 63 additions & 6 deletions src/a2a/server/request_handlers/grpc_handler.py
Original file line number Diff line number Diff line change
Expand Up @@ -3,7 +3,7 @@

from abc import ABC, abstractmethod
from collections.abc import AsyncIterable, Awaitable, Callable
from typing import TypeVar
from typing import Any, TypeVar


try:
Expand All @@ -18,6 +18,7 @@

from google.protobuf import any_pb2, empty_pb2, message
from google.rpc import error_details_pb2, status_pb2
from packaging.version import InvalidVersion, Version

import a2a.types.a2a_pb2_grpc as a2a_grpc

Expand All @@ -30,7 +31,7 @@
from a2a.server.context import ServerCallContext
from a2a.server.request_handlers.request_handler import RequestHandler
from a2a.types import a2a_pb2
from a2a.utils import proto_utils
from a2a.utils import constants, 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
Expand All @@ -52,7 +53,19 @@ class DefaultGrpcServerCallContextBuilder(GrpcServerCallContextBuilder):

def build(self, context: grpc.aio.ServicerContext) -> ServerCallContext:
"""Builds a ServerCallContext from a gRPC ServicerContext."""
state = {'grpc_context': context}
state: dict[str, Any] = {'grpc_context': context}
# Mirror what the HTTP route builders do (`state['headers'] =
# dict(request.headers)`, see a2a.server.routes.common): helpers such
# as `validate_version` read the A2A-Version header out of
# `context.state['headers']`. Without this key the gRPC transport has
# no way to see request metadata at all. Keys are lowercased because
# gRPC metadata keys are case-insensitive and normalized to lowercase.
state['headers'] = {
key.lower(): (
value.decode('utf-8') if isinstance(value, bytes) else value
)
for key, value in (context.invocation_metadata() or ())
}
return ServerCallContext(
user=self.build_user(context),
state=state,
Expand Down Expand Up @@ -88,16 +101,59 @@ def _get_metadata_value(
types.InternalError: grpc.StatusCode.INTERNAL,
types.TaskNotFoundError: grpc.StatusCode.NOT_FOUND,
types.TaskNotCancelableError: grpc.StatusCode.FAILED_PRECONDITION,
types.PushNotificationNotSupportedError: grpc.StatusCode.FAILED_PRECONDITION,
types.UnsupportedOperationError: grpc.StatusCode.FAILED_PRECONDITION,
types.PushNotificationNotSupportedError: grpc.StatusCode.UNIMPLEMENTED,
types.UnsupportedOperationError: grpc.StatusCode.UNIMPLEMENTED,
types.ContentTypeNotSupportedError: grpc.StatusCode.INVALID_ARGUMENT,
types.InvalidAgentResponseError: grpc.StatusCode.INTERNAL,
types.ExtendedAgentCardNotConfiguredError: grpc.StatusCode.FAILED_PRECONDITION,
types.ExtensionSupportRequiredError: grpc.StatusCode.FAILED_PRECONDITION,
types.VersionNotSupportedError: grpc.StatusCode.FAILED_PRECONDITION,
types.VersionNotSupportedError: grpc.StatusCode.UNIMPLEMENTED,
}


def _validate_a2a_version(server_context: ServerCallContext) -> None:
"""Rejects requests whose A2A-Version metadata this handler can't serve.

The JSON-RPC and REST dispatchers already enforce this through the
`@validate_version(PROTOCOL_VERSION_1_0)` decorator; the gRPC transport had
no equivalent, so `A2A-Version: 99.0` was processed as if it were 1.0.
Spec §7 (Version Negotiation): "If the version is not supported by the
interface, agents MUST return a `VersionNotSupportedError`."

Applied here — in the one place every RPC funnels through — rather than as
a decorator on each of the 11 servicer methods. `VersionNotSupportedError`
is an `A2AError`, so the existing `except A2AError` in `_handle_unary` /
`_handle_stream` maps it to a gRPC status via `_ERROR_CODE_MAP`.
"""
headers = server_context.state.get('headers', {})
actual = headers.get(constants.VERSION_HEADER.lower()) or headers.get(
constants.VERSION_HEADER
)
if not actual:
# Same default as `validate_version`: absent header means 0.3.
actual = constants.PROTOCOL_VERSION_0_3
actual = str(actual)
if actual == constants.PROTOCOL_VERSION_1_0:
return
try:
compatible = (
Version(actual).major
== Version(constants.PROTOCOL_VERSION_1_0).major
)
except InvalidVersion:
compatible = False
if not compatible:
logger.warning(
"Version mismatch: actual='%s', expected='%s'",
actual,
constants.PROTOCOL_VERSION_1_0,
)
raise types.VersionNotSupportedError(
message=f"A2A version '{actual}' is not supported by this handler. "
f"Expected version '{constants.PROTOCOL_VERSION_1_0}'."
)


TResponse = TypeVar('TResponse')


Expand Down Expand Up @@ -425,4 +481,5 @@ def _build_call_context(
) -> ServerCallContext:
server_context = self._context_builder.build(context)
server_context.tenant = getattr(request, 'tenant', '')
_validate_a2a_version(server_context)
return server_context
15 changes: 13 additions & 2 deletions tests/server/request_handlers/test_grpc_handler.py
Original file line number Diff line number Diff line change
Expand Up @@ -26,6 +26,14 @@ def mock_grpc_context() -> AsyncMock:
context = AsyncMock(spec=grpc.aio.ServicerContext)
context.abort = AsyncMock()
context.set_trailing_metadata = MagicMock()
# Mirrors the JSON-RPC/REST test fixtures, which default their TestClient
# to `headers={'A2A-Version': '1.0'}` (see test_jsonrpc_dispatcher.py,
# test_rest_dispatcher.py) now that GrpcHandler validates A2A-Version the
# same way those dispatchers already did. Tests exercising version
# negotiation itself override `invocation_metadata.return_value`.
context.invocation_metadata.return_value = grpc.aio.Metadata(
('a2a-version', '1.0'),
)
return context


Expand Down Expand Up @@ -334,12 +342,12 @@ async def test_list_tasks_success(
),
(
types.PushNotificationNotSupportedError(),
grpc.StatusCode.FAILED_PRECONDITION,
grpc.StatusCode.UNIMPLEMENTED,
'PushNotificationNotSupportedError',
),
(
types.UnsupportedOperationError(),
grpc.StatusCode.FAILED_PRECONDITION,
grpc.StatusCode.UNIMPLEMENTED,
'UnsupportedOperationError',
),
(
Expand Down Expand Up @@ -417,6 +425,7 @@ async def test_send_message_with_extensions(
mock_grpc_context: AsyncMock,
) -> None:
mock_grpc_context.invocation_metadata.return_value = grpc.aio.Metadata(
('a2a-version', '1.0'),
(HTTP_EXTENSION_HEADER.lower(), 'foo'),
(HTTP_EXTENSION_HEADER.lower(), 'bar'),
)
Expand All @@ -442,6 +451,7 @@ async def test_send_message_with_comma_separated_extensions(
mock_grpc_context: AsyncMock,
) -> None:
mock_grpc_context.invocation_metadata.return_value = grpc.aio.Metadata(
('a2a-version', '1.0'),
(HTTP_EXTENSION_HEADER.lower(), 'foo ,, bar,'),
(HTTP_EXTENSION_HEADER.lower(), 'baz , bar'),
)
Expand All @@ -467,6 +477,7 @@ async def test_send_streaming_message_with_extensions(
mock_grpc_context: AsyncMock,
) -> None:
mock_grpc_context.invocation_metadata.return_value = grpc.aio.Metadata(
('a2a-version', '1.0'),
(HTTP_EXTENSION_HEADER.lower(), 'foo'),
(HTTP_EXTENSION_HEADER.lower(), 'bar'),
)
Expand Down