Skip to content

feat(grpc): Add util method to extract Task ID and Push Notification Config ID from request #524

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
wants to merge 4 commits into
base: main
Choose a base branch
from
Open

feat(grpc): Add util method to extract Task ID and Push Notification Config ID from request #524

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
084aca1
Add util method to extract Task ID and Push Notification Config ID fr…
rinaldiangoogle Oct 27, 2025
56e7448
Add docstring for `task_push_notification_config_params` method
rinaldiangoogle Oct 27, 2025
b4fef7f
Format
holtskinner Oct 28, 2025
65cdf3d
Add tests for valid and invalid calls to
rinaldiangoogle Oct 28, 2025
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
31 changes: 31 additions & 0 deletions src/a2a/utils/proto_utils.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -366,6 +366,37 @@ def task_push_notification_config(
),
)

@classmethod
def task_push_notification_config_params(
Copy link
Member

@lkawka lkawka Oct 31, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I think what would be more useful is a generic function to parse PushNotifiactionConfig's name from a string. Such a function could be used in various contexts beyond these specific requests.

Copy link
Author

@rinaldiangoogle rinaldiangoogle Nov 3, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I guess an option could be to get a string as parameter and fail if not in the tasks/{task_id}/pushNotificationConfigs/{config_id} format. Although, as in the PR description, that would still not be possible to have a more flexible function and output a PushNotifiactionConfig since there the url property is required (unless we change that).
Since we are in the proto_utils I think it is fair to working with these proto requests as parameters.

Copy link
Member

@lkawka lkawka Nov 4, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The behavior would be the similar, your current implementation also fails if the format is incorrect, and you can always move the function to a different utils file. I just think it would be more versatile to have it accept a string value.

I feel like returning the PushNotifiactionConfig would be misleading here. You cannot build it from GetTaskPushNotificationConfigRequest.

cls,
request: (
a2a_pb2.GetTaskPushNotificationConfigRequest
| a2a_pb2.DeleteTaskPushNotificationConfigRequest
),
) -> tuple[str, str]:
"""Parses the task ID and push notification config ID from the request.

Args:
request: The request to parse.

Returns:
A tuple containing the task ID and push notification config ID.

Raises:
ServerError: If the request name is invalid.
"""
m = _TASK_PUSH_CONFIG_NAME_MATCH.match(request.name)
if not m:
raise ServerError(
error=types.InvalidParamsError(
message=f'No task or config id for {request.name}'
)
)
return (
m.group(1),
m.group(2),
)

@classmethod
def agent_card(
cls,
Expand Down
56 changes: 56 additions & 0 deletions tests/utils/test_proto_utils.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -244,6 +244,62 @@ def test_task_push_config_from_proto_invalid_parent(self):
proto_utils.FromProto.task_push_notification_config(request)
assert isinstance(exc_info.value.error, types.InvalidParamsError)

@pytest.mark.parametrize(
'request_type, name, expected_task_id, expected_config_id',
[
(
a2a_pb2.GetTaskPushNotificationConfigRequest,
'tasks/task-123/pushNotificationConfigs/config-456',
'task-123',
'config-456',
),
(
a2a_pb2.DeleteTaskPushNotificationConfigRequest,
'tasks/task-abc/pushNotificationConfigs/config-def',
'task-abc',
'config-def',
),
],
)
def test_task_push_notification_config_params_valid(
self, request_type, name, expected_task_id, expected_config_id
):
"""Test valid name in task_push_notification_config_params."""
request = request_type(name=name)
task_id, config_id = (
proto_utils.ToProto.task_push_notification_config_params(request)
)
assert task_id == expected_task_id
assert config_id == expected_config_id

@pytest.mark.parametrize(
'request_type',
[
a2a_pb2.GetTaskPushNotificationConfigRequest,
a2a_pb2.DeleteTaskPushNotificationConfigRequest,
],
)
@pytest.mark.parametrize(
'name',
[
'invalid-name-format',
'tasks/task-123',
'tasks/task-123/pushNotificationConfigs',
'tasks//pushNotificationConfigs/config-456',
'tasks/task-123/pushNotificationConfigs/',
'foo/task-123/bar/config-456',
],
)
def test_task_push_notification_config_params_invalid(
self, request_type, name
):
"""Test invalid names in task_push_notification_config_params."""
request = request_type(name=name)
with pytest.raises(ServerError) as exc_info:
proto_utils.ToProto.task_push_notification_config_params(request)
assert isinstance(exc_info.value.error, types.InvalidParamsError)
assert 'No task or config id for' in exc_info.value.error.message

def test_none_handling(self):
"""Test that None inputs are handled gracefully."""
assert proto_utils.ToProto.message(None) is None
Expand Down
Loading