Skip to content
Merged
Show file tree
Hide file tree
Changes from 1 commit
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
226 changes: 226 additions & 0 deletions src/a2a/helpers/proto_helpers.py
Original file line number Diff line number Diff line change
Expand Up @@ -3,6 +3,10 @@
import uuid

from collections.abc import Sequence
from typing import Any

from google.protobuf import struct_pb2
from google.protobuf.json_format import ParseDict

from a2a.types.a2a_pb2 import (
Artifact,
Expand Down Expand Up @@ -57,6 +61,89 @@ def get_message_text(message: Message, delimiter: str = '\n') -> str:
return delimiter.join(get_text_parts(message.parts))


def new_data_message(
data: Any,
role: Role = Role.ROLE_AGENT,
context_id: str | None = None,
task_id: str | None = None,
) -> Message:
"""Creates a new message containing a single data Part.

Args:
data: JSON-serializable data to embed (dict, list, str, etc.).
role: The role of the message sender (default: ROLE_AGENT).
context_id: Optional context ID.
task_id: Optional task ID.

Returns:
A Message with a single data Part.
"""
return new_message(
parts=[new_data_part(data)],
role=role,
context_id=context_id,
task_id=task_id,
)


def new_raw_message( # noqa: PLR0913
raw: bytes,
media_type: str | None = None,
filename: str | None = None,
role: Role = Role.ROLE_AGENT,
context_id: str | None = None,
task_id: str | None = None,
) -> Message:
"""Creates a new message containing a single raw bytes Part.

Args:
raw: The raw bytes content.
media_type: Optional MIME type (e.g. 'image/png').
filename: Optional filename.
role: The role of the message sender (default: ROLE_AGENT).
context_id: Optional context ID.
task_id: Optional task ID.

Returns:
A Message with a single raw Part.
"""
return new_message(
parts=[new_raw_part(raw, media_type=media_type, filename=filename)],
role=role,
context_id=context_id,
task_id=task_id,
)


def new_url_message( # noqa: PLR0913
url: str,
media_type: str | None = None,
filename: str | None = None,
role: Role = Role.ROLE_AGENT,
context_id: str | None = None,
task_id: str | None = None,
) -> Message:
"""Creates a new message containing a single URL Part.

Args:
url: The URL pointing to the file content.
media_type: Optional MIME type (e.g. 'image/png').
filename: Optional filename.
role: The role of the message sender (default: ROLE_AGENT).
context_id: Optional context ID.
task_id: Optional task ID.

Returns:
A Message with a single URL Part.
"""
return new_message(
parts=[new_url_part(url, media_type=media_type, filename=filename)],
role=role,
context_id=context_id,
task_id=task_id,
)


# --- Artifact Helpers ---


Expand Down Expand Up @@ -90,6 +177,89 @@ def new_text_artifact(
)


def new_data_artifact(
name: str,
data: Any,
description: str | None = None,
artifact_id: str | None = None,
) -> Artifact:
"""Creates a new Artifact object containing only a single data Part.

Args:
name: The name of the artifact.
data: JSON-serializable data to embed (dict, list, str, etc.).
description: Optional description.
artifact_id: Optional artifact ID (auto-generated if not provided).

Returns:
An Artifact with a single data Part.
"""
return new_artifact(
[new_data_part(data)],
name,
description,
artifact_id=artifact_id,
)


def new_raw_artifact( # noqa: PLR0913
name: str,
raw: bytes,
media_type: str | None = None,
filename: str | None = None,
description: str | None = None,
artifact_id: str | None = None,
) -> Artifact:
"""Creates a new Artifact object containing only a single raw bytes Part.

Args:
name: The name of the artifact.
raw: The raw bytes content.
media_type: Optional MIME type (e.g. 'image/png').
filename: Optional filename.
description: Optional description.
artifact_id: Optional artifact ID (auto-generated if not provided).

Returns:
An Artifact with a single raw Part.
"""
return new_artifact(
[new_raw_part(raw, media_type=media_type, filename=filename)],
name,
description,
artifact_id=artifact_id,
)


def new_url_artifact( # noqa: PLR0913
name: str,
url: str,
media_type: str | None = None,
filename: str | None = None,
description: str | None = None,
artifact_id: str | None = None,
) -> Artifact:
"""Creates a new Artifact object containing only a single URL Part.

Args:
name: The name of the artifact.
url: The URL pointing to the file content.
media_type: Optional MIME type (e.g. 'image/png').
filename: Optional filename.
description: Optional description.
artifact_id: Optional artifact ID (auto-generated if not provided).

Returns:
An Artifact with a single URL Part.
"""
return new_artifact(
[new_url_part(url, media_type=media_type, filename=filename)],
name,
description,
artifact_id=artifact_id,
)


def get_artifact_text(artifact: Artifact, delimiter: str = '\n') -> str:
"""Extracts and joins all text content from an Artifact's parts."""
return delimiter.join(get_text_parts(artifact.parts))
Expand Down Expand Up @@ -141,6 +311,62 @@ def new_task(
# --- Part Helpers ---


def new_data_part(data: Any) -> Part:
"""Creates a Part with structured data (google.protobuf.Value).

Args:
data: JSON-serializable data to embed (dict, list, str, etc.).

Returns:
A Part with the data field set.
"""
return Part(data=ParseDict(data, struct_pb2.Value()))
Comment thread
guglielmo-san marked this conversation as resolved.
Outdated


def new_raw_part(
raw: bytes,
media_type: str | None = None,
filename: str | None = None,
) -> Part:
"""Creates a Part with raw bytes content.

Args:
raw: The raw bytes content.
media_type: Optional MIME type (e.g. 'image/png').
filename: Optional filename.

Returns:
A Part with the raw field set.
"""
return Part(
raw=raw,
media_type=media_type or '',
filename=filename or '',
)


def new_url_part(
url: str,
media_type: str | None = None,
filename: str | None = None,
) -> Part:
"""Creates a Part with a URL pointing to file content.

Args:
url: The URL to the file content.
media_type: Optional MIME type (e.g. 'image/png').
filename: Optional filename.

Returns:
A Part with the url field set.
"""
return Part(
url=url,
media_type=media_type or '',
filename=filename or '',
)


def get_text_parts(parts: Sequence[Part]) -> list[str]:
"""Extracts text content from all text Parts."""
return [part.text for part in parts if part.HasField('text')]
Expand Down
Loading
Loading