Skip to content
89 changes: 75 additions & 14 deletions src/a2a/utils/proto_utils.py
Original file line number Diff line number Diff line change
@@ -1,3 +1,3 @@
# mypy: disable-error-code="arg-type"
"""Utils for converting between proto and Python types."""

Expand Down Expand Up @@ -46,14 +46,7 @@
) -> struct_pb2.Struct | None:
if metadata is None:
return None
return struct_pb2.Struct(
# TODO: Add support for other types.
fields={
key: struct_pb2.Value(string_value=value)
for key, value in metadata.items()
if isinstance(value, str)
}
)
return dict_to_struct(metadata)

@classmethod
def part(cls, part: types.Part) -> a2a_pb2.Part:
Expand Down Expand Up @@ -81,7 +74,9 @@
) -> a2a_pb2.FilePart:
if isinstance(file, types.FileWithUri):
return a2a_pb2.FilePart(
file_with_uri=file.uri, mime_type=file.mime_type, name=file.name
file_with_uri=file.uri,
mime_type=file.mime_type,
name=file.name
)
return a2a_pb2.FilePart(
file_with_bytes=file.bytes.encode('utf-8'),
Expand Down Expand Up @@ -324,6 +319,23 @@
return a2a_pb2.AgentCapabilities(
streaming=bool(capabilities.streaming),
push_notifications=bool(capabilities.push_notifications),
extensions=[
cls.extension(x) for x in capabilities.extensions
]
if capabilities.extensions
else None,
Comment thread
pstephengoogle marked this conversation as resolved.
Outdated
)

@classmethod
def extension(
cls,
extension: types.AgentExtension,
) -> a2a_pb2.AgentExtension:
return a2a_pb2.AgentExtension(
uri=extension.uri,
description=extension.description,
params=dict_to_struct(extension.params),

Check failure on line 337 in src/a2a/utils/proto_utils.py

View workflow job for this annotation

GitHub Actions / Lint Code Base

Argument of type "dict[str, Any] | None" cannot be assigned to parameter "dictionary" of type "dict[str, Any]" in function "dict_to_struct"   Type "dict[str, Any] | None" is not assignable to type "dict[str, Any]"     "None" is not assignable to "dict[str, Any]" (reportArgumentType)
required=extension.required,
)

@classmethod
Expand Down Expand Up @@ -477,11 +489,9 @@

@classmethod
def metadata(cls, metadata: struct_pb2.Struct) -> dict[str, Any]:
return {
key: value.string_value
for key, value in metadata.fields.items()
if value.string_value
}
if not metadata:
return {}
return struct_to_dict(metadata)
Comment thread
pstephengoogle marked this conversation as resolved.

@classmethod
def part(cls, part: a2a_pb2.Part) -> types.Part:
Expand Down Expand Up @@ -777,6 +787,23 @@
return types.AgentCapabilities(
streaming=capabilities.streaming,
push_notifications=capabilities.push_notifications,
extensions=[
cls.agent_extension(x) for x in capabilities.extensions
]
if capabilities.extensions
else None,
Comment thread
pstephengoogle marked this conversation as resolved.
Outdated
)

@classmethod
def agent_extension(
cls,
extension: a2a_pb2.AgentExtension,
) -> types.AgentExtension:
return types.AgentExtension(
uri=extension.uri,
description=extension.description,
params=struct_to_dict(extension.params),
required=extension.required,
)

@classmethod
Expand Down Expand Up @@ -916,3 +943,37 @@
return types.Role.agent
case _:
return types.Role.agent


def struct_to_dict(struct: struct_pb2.Struct) -> dict[str, Any]:
"""Converts a Struct proto to a Python dict."""

def convert(value: struct_pb2.Value) -> Any:
if value.HasField('list_value'):
return [convert(v) for v in value.list_value.values]
elif value.HasField('struct_value'):

Check failure on line 954 in src/a2a/utils/proto_utils.py

View workflow job for this annotation

GitHub Actions / Lint Code Base

Ruff (RET505)

src/a2a/utils/proto_utils.py:954:5: RET505 Unnecessary `elif` after `return` statement
return {k: convert(v) for k, v in value.struct_value.fields.items()}
elif value.HasField('number_value'):
return value.number_value
elif value.HasField('string_value'):
return value.string_value
elif value.HasField('bool_value'):
return value.bool_value
elif value.HasField('null_value'):
return None
else:
raise ValueError(f'Unsupported type: {value}')

return {k: convert(v) for k, v in struct.fields.items()}
Comment thread
pstephengoogle marked this conversation as resolved.
Outdated


def dict_to_struct(dictionary: dict[str, Any]) -> struct_pb2.Struct:
"""Converts a Python dict to a Struct proto."""
struct = struct_pb2.Struct()
for key, val in dictionary.items():
if isinstance(val, dict):
struct[key] = dict_to_struct(val)
else:
struct[key] = val
return struct
Comment thread
pstephengoogle marked this conversation as resolved.
Outdated

Loading