fix: validate task ID length and message content

src/a2a/server/request_handlers/default_request_handler.py

@@ -61,7 +61,9 @@
 from a2a.utils.task import (
     apply_history_length,
     validate_history_length,
+    validate_message_content,
     validate_page_size,
+    validate_task_id,
 )
 from a2a.utils.telemetry import SpanKind, trace_class
 
@@ -145,18 +147,19 @@
         context: ServerCallContext,
     ) -> Task | None:
         """Default handler for 'tasks/get'."""
         validate_history_length(params)
+        validate_task_id(params.id)
 
         task_id = params.id
         task: Task | None = await self.task_store.get(task_id, context)
         if not task:
             raise TaskNotFoundError
 
         return apply_history_length(task, params)
 
     @validate_request_params
     async def on_list_tasks(
         self,
         params: ListTasksRequest,
         context: ServerCallContext,
     ) -> ListTasksResponse:
@@ -187,6 +190,7 @@
         Attempts to cancel the task managed by the `AgentExecutor`.
         """
         task_id = params.id
+        validate_task_id(task_id)
         task: Task | None = await self.task_store.get(task_id, context)
         if not task:
             raise TaskNotFoundError
@@ -265,6 +269,9 @@
         # Create task manager and validate existing task
         # Proto empty strings should be treated as None
         task_id = params.message.task_id or None
+        if task_id:
+            validate_task_id(task_id)
+        validate_message_content(params.message)
         context_id = params.message.context_id or None
         task_manager = TaskManager(
             task_id=task_id,
@@ -519,30 +526,31 @@
 
         Requires a `PushNotifier` to be configured.
         """
         if not self._push_config_store:
             raise PushNotificationNotSupportedError
 
         task_id = params.task_id
+        validate_task_id(task_id)
         task: Task | None = await self.task_store.get(task_id, context)
         if not task:
             raise TaskNotFoundError
 
         await self._push_config_store.set_info(
             task_id,
             params,
             context,
         )
 
         return params
 
     @validate_request_params
     @validate(
         lambda self: self._agent_card.capabilities.push_notifications,
         error_message='Push notifications are not supported by the agent',
         error_type=PushNotificationNotSupportedError,
     )
     async def on_get_task_push_notification_config(
         self,
         params: GetTaskPushNotificationConfigRequest,
         context: ServerCallContext,
     ) -> TaskPushNotificationConfig:
@@ -550,32 +558,33 @@
 
         Requires a `PushConfigStore` to be configured.
         """
         if not self._push_config_store:
             raise PushNotificationNotSupportedError
 
         task_id = params.task_id
         config_id = params.id
+        validate_task_id(task_id)
         task: Task | None = await self.task_store.get(task_id, context)
         if not task:
             raise TaskNotFoundError
 
         push_notification_configs: list[TaskPushNotificationConfig] = (
             await self._push_config_store.get_info(task_id, context) or []
         )
 
         for config in push_notification_configs:
             if config.id == config_id:
                 return config
 
         raise TaskNotFoundError
 
     @validate_request_params
     @validate(
         lambda self: self._agent_card.capabilities.streaming,
         'Streaming is not supported by the agent',
     )
     async def on_subscribe_to_task(
         self,
         params: SubscribeToTaskRequest,
         context: ServerCallContext,
     ) -> AsyncGenerator[Event, None]:
@@ -585,6 +594,7 @@
         Requires the task and its queue to still be active.
         """
         task_id = params.id
+        validate_task_id(task_id)
         task: Task | None = await self.task_store.get(task_id, context)
         if not task:
             raise TaskNotFoundError
@@ -631,30 +641,31 @@
 
         Requires a `PushConfigStore` to be configured.
         """
         if not self._push_config_store:
             raise PushNotificationNotSupportedError
 
         task_id = params.task_id
+        validate_task_id(task_id)
         task: Task | None = await self.task_store.get(task_id, context)
         if not task:
             raise TaskNotFoundError
 
         push_notification_config_list = await self._push_config_store.get_info(
             task_id, context
         )
 
         return ListTaskPushNotificationConfigsResponse(
             configs=push_notification_config_list
         )
 
     @validate_request_params
     @validate(
         lambda self: self._agent_card.capabilities.push_notifications,
         error_message='Push notifications are not supported by the agent',
         error_type=PushNotificationNotSupportedError,
     )
     async def on_delete_task_push_notification_config(
         self,
         params: DeleteTaskPushNotificationConfigRequest,
         context: ServerCallContext,
     ) -> None:
@@ -662,38 +673,39 @@
 
         Requires a `PushConfigStore` to be configured.
         """
         if not self._push_config_store:
             raise PushNotificationNotSupportedError
 
         task_id = params.task_id
         config_id = params.id
+        validate_task_id(task_id)
         task: Task | None = await self.task_store.get(task_id, context)
         if not task:
             raise TaskNotFoundError
 
         await self._push_config_store.delete_info(task_id, context, config_id)
 
     @validate_request_params
     @validate(
         lambda self: self._agent_card.capabilities.extended_agent_card,
         error_message='The agent does not support authenticated extended cards',
     )
     async def on_get_extended_agent_card(
         self,
         params: GetExtendedAgentCardRequest,
         context: ServerCallContext,
     ) -> AgentCard:
         """Default handler for 'GetExtendedAgentCard'.
 
         Requires `capabilities.extended_agent_card` to be true.
         """
         extended_card = self.extended_agent_card
         if not extended_card:
             raise ExtendedAgentCardNotConfiguredError
 
         if self.extended_card_modifier:
             extended_card = await self.extended_card_modifier(
                 extended_card, context
             )
 
         return extended_card

src/a2a/server/request_handlers/default_request_handler_v2.py

@@ -49,7 +49,9 @@
 from a2a.utils.task import (
     apply_history_length,
     validate_history_length,
+    validate_message_content,
     validate_page_size,
+    validate_task_id,
 )
 from a2a.utils.telemetry import SpanKind, trace_class
 
@@ -127,17 +129,18 @@
         params: GetTaskRequest,
         context: ServerCallContext,
     ) -> Task | None:
         validate_history_length(params)
+        validate_task_id(params.id)
 
         task_id = params.id
         task: Task | None = await self.task_store.get(task_id, context)
         if not task:
             raise TaskNotFoundError
 
         return apply_history_length(task, params)
 
     @validate_request_params
     async def on_list_tasks(  # noqa: D102
         self,
         params: ListTasksRequest,
         context: ServerCallContext,
@@ -164,6 +167,7 @@
         context: ServerCallContext,
     ) -> Task | None:
         task_id = params.id
+        validate_task_id(task_id)
 
         try:
             active_task = await self._active_task_registry.get_or_create(
@@ -197,6 +201,9 @@
         validate_history_length(params.configuration)
 
         original_task_id = params.message.task_id or None
+        if original_task_id:
+            validate_task_id(original_task_id)
+        validate_message_content(params.message)
         original_context_id = params.message.context_id or None
 
         if original_task_id:
@@ -337,63 +344,66 @@
         params: TaskPushNotificationConfig,
         context: ServerCallContext,
     ) -> TaskPushNotificationConfig:
         if not self._push_config_store:
             raise PushNotificationNotSupportedError
 
         task_id = params.task_id
+        validate_task_id(task_id)
         task: Task | None = await self.task_store.get(task_id, context)
         if not task:
             raise TaskNotFoundError
 
         await self._push_config_store.set_info(
             task_id,
             params,
             context,
         )
 
         return params
 
     @validate_request_params
     @validate(
         lambda self: self._agent_card.capabilities.push_notifications,
         error_message='Push notifications are not supported by the agent',
         error_type=PushNotificationNotSupportedError,
     )
     async def on_get_task_push_notification_config(  # noqa: D102
         self,
         params: GetTaskPushNotificationConfigRequest,
         context: ServerCallContext,
     ) -> TaskPushNotificationConfig:
         if not self._push_config_store:
             raise PushNotificationNotSupportedError
 
         task_id = params.task_id
         config_id = params.id
+        validate_task_id(task_id)
         task: Task | None = await self.task_store.get(task_id, context)
         if not task:
             raise TaskNotFoundError
 
         push_notification_configs: list[TaskPushNotificationConfig] = (
             await self._push_config_store.get_info(task_id, context) or []
         )
 
         for config in push_notification_configs:
             if config.id == config_id:
                 return config
 
         raise TaskNotFoundError
 
     @validate_request_params
     @validate(
         lambda self: self._agent_card.capabilities.streaming,
         'Streaming is not supported by the agent',
     )
     async def on_subscribe_to_task(  # noqa: D102
         self,
         params: SubscribeToTaskRequest,
         context: ServerCallContext,
     ) -> AsyncGenerator[Event, None]:
         task_id = params.id
+        validate_task_id(task_id)
 
         active_task = await self._active_task_registry.get_or_create(
             task_id,
@@ -415,65 +425,67 @@
         params: ListTaskPushNotificationConfigsRequest,
         context: ServerCallContext,
     ) -> ListTaskPushNotificationConfigsResponse:
         if not self._push_config_store:
             raise PushNotificationNotSupportedError
 
         task_id = params.task_id
+        validate_task_id(task_id)
         task: Task | None = await self.task_store.get(task_id, context)
         if not task:
             raise TaskNotFoundError
 
         push_notification_config_list = await self._push_config_store.get_info(
             task_id, context
         )
 
         return ListTaskPushNotificationConfigsResponse(
             configs=push_notification_config_list
         )
 
     @validate_request_params
     @validate(
         lambda self: self._agent_card.capabilities.push_notifications,
         error_message='Push notifications are not supported by the agent',
         error_type=PushNotificationNotSupportedError,
     )
     async def on_delete_task_push_notification_config(  # noqa: D102
         self,
         params: DeleteTaskPushNotificationConfigRequest,
         context: ServerCallContext,
     ) -> None:
         if not self._push_config_store:
             raise PushNotificationNotSupportedError
 
         task_id = params.task_id
         config_id = params.id
+        validate_task_id(task_id)
         task: Task | None = await self.task_store.get(task_id, context)
         if not task:
             raise TaskNotFoundError
 
         await self._push_config_store.delete_info(task_id, context, config_id)
 
     @validate_request_params
     @validate(
         lambda self: self._agent_card.capabilities.extended_agent_card,
         error_message='The agent does not support authenticated extended cards',
     )
     async def on_get_extended_agent_card(
         self,
         params: GetExtendedAgentCardRequest,
         context: ServerCallContext,
     ) -> AgentCard:
         """Default handler for 'GetExtendedAgentCard'.
 
         Requires `capabilities.extended_agent_card` to be true.
         """
         extended_card = self.extended_agent_card
         if not extended_card:
             raise ExtendedAgentCardNotConfiguredError
 
         if self.extended_card_modifier:
             extended_card = await self.extended_card_modifier(
                 extended_card, context
             )
 
         return extended_card

src/a2a/utils/task.py

@@ -5,11 +5,65 @@
 from base64 import b64decode, b64encode
 from typing import Literal, Protocol, runtime_checkable
 
-from a2a.types.a2a_pb2 import Task
+from a2a.types.a2a_pb2 import Message, Part, Task
 from a2a.utils.constants import MAX_LIST_TASKS_PAGE_SIZE
 from a2a.utils.errors import InvalidParamsError
 
 
+MAX_TASK_ID_LENGTH = 1000
+"""Maximum allowed length of a task ID."""
+
+
+def validate_task_id(task_id: str) -> None:
+    """Validates that a task ID is non-empty and within the length limit.
+
+    Raises:
+        InvalidParamsError: If the task ID is empty or longer than
+            ``MAX_TASK_ID_LENGTH`` characters.
+    """
+    if not task_id:
+        raise InvalidParamsError(message='task ID must be non-empty')
+    if len(task_id) > MAX_TASK_ID_LENGTH:
+        raise InvalidParamsError(
+            message=f'task ID must be at most {MAX_TASK_ID_LENGTH} characters'
+        )
+
+
+def validate_message_content(message: Message) -> None:
+    """Validates that a message carries actual content.
+
+    A message must contain at least one part, and every part must have
+    content (text, raw bytes, a URL or data) rather than being empty.
+
+    Raises:
+        InvalidParamsError: If the message has no parts or contains an
+            empty part.
+    """
+    if not message.parts:
+        raise InvalidParamsError(
+            message='message must contain at least one part'
+        )
+    for part in message.parts:
+        if not _part_has_content(part):
+            raise InvalidParamsError(message='message parts must not be empty')
+
+
+def _part_has_content(part: Part) -> bool:
+    """Returns True if a part carries actual content.
+
+    A part has content if it has text, raw bytes or a URL, or a
+    non-null ``data`` payload (``google.protobuf.Value`` defaults to
+    null/empty).
+    """
+    if part.text or part.raw or part.url:
+        return True
+    return (
+        part.data.WhichOneof('kind') not in (None, 'null_value')
+        if part.HasField('data')
+        else False
+    )
+
+
 @runtime_checkable
 class HistoryLengthConfig(Protocol):
     """Protocol for configuration arguments containing history_length field."""

tests/server/request_handlers/test_default_request_handler.py

@@ -3143,3 +3143,107 @@ async def test_on_get_task_push_notification_config_is_owner_scoped(
             ),
             _ctx('bob'),
         )
+
+
+@pytest.mark.asyncio
+async def test_on_get_task_overlong_task_id_error(agent_card):
+    """A task ID longer than the limit is rejected with InvalidParamsError."""
+    mock_task_store = AsyncMock(spec=TaskStore)
+    request_handler = DefaultRequestHandler(
+        agent_executor=AsyncMock(spec=AgentExecutor),
+        task_store=mock_task_store,
+        agent_card=agent_card,
+    )
+    params = GetTaskRequest(id='a' * 1001)
+    context = create_server_call_context()
+
+    with pytest.raises(InvalidParamsError):
+        await request_handler.on_get_task(params, context)
+    mock_task_store.get.assert_not_awaited()
+
+
+@pytest.mark.asyncio
+async def test_on_cancel_task_overlong_task_id_error(agent_card):
+    """A task ID longer than the limit is rejected before touching the store."""
+    mock_task_store = AsyncMock(spec=TaskStore)
+    request_handler = DefaultRequestHandler(
+        agent_executor=AsyncMock(spec=AgentExecutor),
+        task_store=mock_task_store,
+        agent_card=agent_card,
+    )
+    params = CancelTaskRequest(id='a' * 1001)
+    context = create_server_call_context()
+
+    with pytest.raises(InvalidParamsError):
+        await request_handler.on_cancel_task(params, context)
+    mock_task_store.get.assert_not_awaited()
+
+
+@pytest.mark.asyncio
+async def test_on_message_send_empty_parts_error(agent_card):
+    """A message with no parts is rejected with InvalidParamsError."""
+    mock_task_store = AsyncMock(spec=TaskStore)
+    request_handler = DefaultRequestHandler(
+        agent_executor=AsyncMock(spec=AgentExecutor),
+        task_store=mock_task_store,
+        agent_card=agent_card,
+    )
+    params = SendMessageRequest(
+        message=Message(
+            role=Role.ROLE_USER,
+            message_id='msg1',
+            parts=[],
+        ),
+    )
+    context = create_server_call_context()
+
+    with pytest.raises(InvalidParamsError):
+        await request_handler.on_message_send(params, context)
+    mock_task_store.get.assert_not_awaited()
+
+
+@pytest.mark.asyncio
+async def test_on_message_send_empty_part_error(agent_card):
+    """A message whose only part is empty is rejected with InvalidParamsError."""
+    mock_task_store = AsyncMock(spec=TaskStore)
+    request_handler = DefaultRequestHandler(
+        agent_executor=AsyncMock(spec=AgentExecutor),
+        task_store=mock_task_store,
+        agent_card=agent_card,
+    )
+    params = SendMessageRequest(
+        message=Message(
+            role=Role.ROLE_USER,
+            message_id='msg1',
+            parts=[Part(media_type='text/plain')],
+        ),
+    )
+    context = create_server_call_context()
+
+    with pytest.raises(InvalidParamsError) as exc_info:
+        await request_handler.on_message_send(params, context)
+
+    assert 'not be empty' in exc_info.value.message
+
+
+@pytest.mark.asyncio
+async def test_on_message_send_overlong_task_id_error(agent_card):
+    """A message targeting an over-long task ID is rejected."""
+    mock_task_store = AsyncMock(spec=TaskStore)
+    request_handler = DefaultRequestHandler(
+        agent_executor=AsyncMock(spec=AgentExecutor),
+        task_store=mock_task_store,
+        agent_card=agent_card,
+    )
+    params = SendMessageRequest(
+        message=Message(
+            role=Role.ROLE_USER,
+            message_id='msg1',
+            task_id='a' * 1001,
+            parts=[Part(text='hello')],
+        ),
+    )
+    context = create_server_call_context()
+
+    with pytest.raises(InvalidParamsError):
+        await request_handler.on_message_send(params, context)