feat(server): validate fields presence according to google.api.field_behavior annotations

Author: ishymko

src/a2a/server/request_handlers/default_request_handler.py

@@ -18,7 +18,10 @@
     InMemoryQueueManager,
     QueueManager,
 )
-from a2a.server.request_handlers.request_handler import RequestHandler
+from a2a.server.request_handlers.request_handler import (
+    RequestHandler,
+    validate_request_params,
+)
 from a2a.server.tasks import (
     PushNotificationConfigStore,
     PushNotificationEvent,
@@ -118,6 +121,7 @@ def __init__(  # noqa: PLR0913
         # asyncio tasks and to surface unexpected exceptions.
         self._background_tasks = set()
 
+    @validate_request_params
     async def on_get_task(
         self,
         params: GetTaskRequest,
@@ -133,6 +137,7 @@ async def on_get_task(
 
         return apply_history_length(task, params)
 
+    @validate_request_params
     async def on_list_tasks(
         self,
         params: ListTasksRequest,
@@ -154,6 +159,7 @@ async def on_list_tasks(
 
         return page
 
+    @validate_request_params
     async def on_cancel_task(
         self,
         params: CancelTaskRequest,
@@ -317,6 +323,7 @@ async def _send_push_notification_if_needed(
         ):
             await self._push_sender.send_notification(task_id, event)
 
+    @validate_request_params
     async def on_message_send(
         self,
         params: SendMessageRequest,
@@ -386,6 +393,7 @@ async def push_notification_callback(event: Event) -> None:
 
         return result
 
+    @validate_request_params
     async def on_message_send_stream(
         self,
         params: SendMessageRequest,
@@ -474,6 +482,7 @@ async def _cleanup_producer(
         async with self._running_agents_lock:
             self._running_agents.pop(task_id, None)
 
+    @validate_request_params
     async def on_create_task_push_notification_config(
         self,
         params: TaskPushNotificationConfig,
@@ -499,6 +508,7 @@ async def on_create_task_push_notification_config(
 
         return params
 
+    @validate_request_params
     async def on_get_task_push_notification_config(
         self,
         params: GetTaskPushNotificationConfigRequest,
@@ -530,6 +540,7 @@ async def on_get_task_push_notification_config(
 
         raise InternalError(message='Push notification config not found')
 
+    @validate_request_params
     async def on_subscribe_to_task(
         self,
         params: SubscribeToTaskRequest,
@@ -572,6 +583,7 @@ async def on_subscribe_to_task(
         async for event in result_aggregator.consume_and_emit(consumer):
             yield event
 
+    @validate_request_params
     async def on_list_task_push_notification_configs(
         self,
         params: ListTaskPushNotificationConfigsRequest,
@@ -597,6 +609,7 @@ async def on_list_task_push_notification_configs(
             configs=push_notification_config_list
         )
 
+    @validate_request_params
     async def on_delete_task_push_notification_config(
         self,
         params: DeleteTaskPushNotificationConfigRequest,

src/a2a/server/request_handlers/request_handler.py

@@ -1,5 +1,11 @@
+import functools
+import inspect
+
 from abc import ABC, abstractmethod
-from collections.abc import AsyncGenerator
+from collections.abc import AsyncGenerator, Callable
+from typing import Any
+
+from google.protobuf.message import Message as ProtoMessage
 
 from a2a.server.context import ServerCallContext
 from a2a.server.events.event_queue import Event
@@ -19,6 +25,7 @@
     TaskPushNotificationConfig,
 )
 from a2a.utils.errors import UnsupportedOperationError
+from a2a.utils.proto_utils import validate_proto_required_fields
 
 
 class RequestHandler(ABC):
@@ -218,3 +225,37 @@ async def on_delete_task_push_notification_config(
         Returns:
             None
         """
+
+
+def validate_request_params(method: Callable) -> Callable:
+    """Decorator for RequestHandler methods to validate required fields on incoming requests."""
+    if inspect.isasyncgenfunction(method):
+
+        @functools.wraps(method)
+        async def async_generator_wrapper(
+            self: RequestHandler,
+            params: ProtoMessage,
+            context: ServerCallContext,
+            *args: Any,
+            **kwargs: Any,
+        ) -> AsyncGenerator:
+            if params is not None:
+                validate_proto_required_fields(params)
+            async for item in method(self, params, context, *args, **kwargs):
+                yield item
+
+        return async_generator_wrapper
+
+    @functools.wraps(method)
+    async def async_wrapper(
+        self: RequestHandler,
+        params: ProtoMessage,
+        context: ServerCallContext,
+        *args: Any,
+        **kwargs: Any,
+    ) -> Any:
+        if params is not None:
+            validate_proto_required_fields(params)
+        return await method(self, params, context, *args, **kwargs)
+
+    return async_wrapper

src/a2a/utils/errors.py

@@ -21,9 +21,10 @@ class A2AError(Exception):
     message: str = 'A2A Error'
     data: dict | None = None
 
-    def __init__(self, message: str | None = None):
+    def __init__(self, message: str | None = None, data: dict | None = None):
         if message:
             self.message = message
+        self.data = data
         super().__init__(self.message)
 
 

src/a2a/utils/proto_utils.py

@@ -17,11 +17,15 @@
 This module provides helper functions for common proto type operations.
 """
 
-from typing import TYPE_CHECKING, Any
+from typing import TYPE_CHECKING, Any, TypedDict
 
+from google.api.field_behavior_pb2 import FieldBehavior, field_behavior
+from google.protobuf.descriptor import FieldDescriptor
 from google.protobuf.json_format import ParseDict
 from google.protobuf.message import Message as ProtobufMessage
 
+from a2a.utils.errors import InvalidParamsError
+
 
 if TYPE_CHECKING:
     from starlette.datastructures import QueryParams
@@ -189,3 +193,106 @@ def parse_params(params: QueryParams, message: ProtobufMessage) -> None:
                 processed[k] = parsed_val
 
     ParseDict(processed, message, ignore_unknown_fields=True)
+
+
+class ValidationDetail(TypedDict):
+    """Structured validation error detail."""
+
+    field: str
+    message: str
+
+
+def _check_required_field_violation(
+    msg: ProtobufMessage, field: FieldDescriptor
+) -> ValidationDetail | None:
+    """Check if a required field is missing or invalid."""
+    val = getattr(msg, field.name)
+    if field.is_repeated:
+        if not val:
+            return ValidationDetail(
+                field=field.name,
+                message='Field must contain at least one element.',
+            )
+    elif field.has_presence:
+        if not msg.HasField(field.name):
+            return ValidationDetail(
+                field=field.name, message='Field is required.'
+            )
+    elif val == field.default_value:
+        return ValidationDetail(field=field.name, message='Field is required.')
+    return None
+
+
+def _append_nested_errors(
+    errors: list[ValidationDetail],
+    prefix: str,
+    sub_errs: list[ValidationDetail],
+) -> None:
+    """Format nested validation errors and append to errors list."""
+    for sub in sub_errs:
+        sub_field = sub['field']
+        errors.append(
+            ValidationDetail(
+                field=f'{prefix}.{sub_field}' if sub_field else prefix,
+                message=sub['message'],
+            )
+        )
+
+
+def _recurse_validation(
+    msg: ProtobufMessage, field: FieldDescriptor
+) -> list[ValidationDetail]:
+    """Recurse validation for nested messages and map fields."""
+    errors: list[ValidationDetail] = []
+    if field.type != FieldDescriptor.TYPE_MESSAGE:
+        return errors
+
+    val = getattr(msg, field.name)
+    if not field.is_repeated:
+        if msg.HasField(field.name):
+            sub_errs = _validate_proto_required_fields_internal(val)
+            _append_nested_errors(errors, field.name, sub_errs)
+    elif field.message_type.GetOptions().map_entry:
+        for k, v in val.items():
+            if isinstance(v, ProtobufMessage):
+                sub_errs = _validate_proto_required_fields_internal(v)
+                _append_nested_errors(errors, f'{field.name}[{k}]', sub_errs)
+    else:
+        for i, item in enumerate(val):
+            sub_errs = _validate_proto_required_fields_internal(item)
+            _append_nested_errors(errors, f'{field.name}[{i}]', sub_errs)
+    return errors
+
+
+def _validate_proto_required_fields_internal(
+    msg: ProtobufMessage,
+) -> list[ValidationDetail]:
+    """Internal validation that returns a list of error dictionaries."""
+    desc = msg.DESCRIPTOR
+    errors: list[ValidationDetail] = []
+
+    for field in desc.fields:
+        options = field.GetOptions()
+        if FieldBehavior.REQUIRED in options.Extensions[field_behavior]:
+            violation = _check_required_field_violation(msg, field)
+            if violation:
+                errors.append(violation)
+        errors.extend(_recurse_validation(msg, field))
+    return errors
+
+
+def validate_proto_required_fields(msg: ProtobufMessage) -> None:
+    """Validate that all fields marked as REQUIRED are present on the proto message.
+
+    Args:
+        msg: The Protobuf message to validate.
+
+    Raises:
+        InvalidParamsError: If a required field is missing or empty.
+    """
+    errors = _validate_proto_required_fields_internal(msg)
+
+    if errors:
+        raise InvalidParamsError(
+            message='Validation failed', data={'errors': errors}
+        )