Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
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
265 changes: 260 additions & 5 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 All @@ -23,9 +27,9 @@

def new_message(
parts: list[Part],
role: Role = Role.ROLE_AGENT,
context_id: str | None = None,
task_id: str | None = None,
role: Role = Role.ROLE_AGENT,
) -> Message:
"""Creates a new message containing a list of Parts."""
return Message(
Expand All @@ -39,16 +43,17 @@ def new_message(

def new_text_message(
text: str,
media_type: str | None = None,
context_id: str | None = None,
task_id: str | None = None,
role: Role = Role.ROLE_AGENT,
) -> Message:
"""Creates a new message containing a single text Part."""
return new_message(
parts=[Part(text=text)],
role=role,
task_id=task_id,
parts=[new_text_part(text, media_type=media_type)],
context_id=context_id,
task_id=task_id,
role=role,
)


Expand All @@ -57,6 +62,91 @@ def get_message_text(message: Message, delimiter: str = '\n') -> str:
return delimiter.join(get_text_parts(message.parts))


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

Args:
data: JSON-serializable data to embed (dict, list, str, etc.).
media_type: Optional MIME type of the part content (e.g., "text/plain", "application/json", "image/png").
context_id: Optional context ID.
task_id: Optional task ID.
role: The role of the message sender (default: ROLE_AGENT).

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


def new_raw_message( # noqa: PLR0913
raw: bytes,
media_type: str | None = None,
filename: str | None = None,
context_id: str | None = None,
task_id: str | None = None,
role: Role = Role.ROLE_AGENT,
) -> 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.
context_id: Optional context ID.
task_id: Optional task ID.
role: The role of the message sender (default: ROLE_AGENT).

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


def new_url_message( # noqa: PLR0913
url: str,
media_type: str | None = None,
filename: str | None = None,
context_id: str | None = None,
task_id: str | None = None,
role: Role = Role.ROLE_AGENT,
) -> 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.
context_id: Optional context ID.
task_id: Optional task ID.
role: The role of the message sender (default: ROLE_AGENT).

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


# --- Artifact Helpers ---


Expand All @@ -78,12 +168,98 @@ def new_artifact(
def new_text_artifact(
name: str,
text: str,
media_type: str | None = None,
description: str | None = None,
artifact_id: str | None = None,
) -> Artifact:
"""Creates a new Artifact object containing only a single text Part."""
return new_artifact(
[Part(text=text)],
[new_text_part(text, media_type=media_type)],
name,
description,
artifact_id=artifact_id,
)


def new_data_artifact(
name: str,
data: Any,
media_type: str | None = None,
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.).
media_type: Optional MIME type of the part content (e.g., "text/plain", "application/json", "image/png").
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, media_type=media_type)],
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,
Expand Down Expand Up @@ -141,6 +317,85 @@ def new_task(
# --- Part Helpers ---


def new_text_part(
text: str,
media_type: str | None = None,
) -> Part:
"""Creates a Part with text content.

Args:
text: The text content.
media_type: Optional MIME type (e.g. 'text/plain', 'text/markdown').

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


def new_data_part(
data: Any,
media_type: str | None = None,
) -> Part:
"""Creates a Part with structured data (google.protobuf.Value).

Args:
data: JSON-serializable data to embed (dict, list, str, etc.).
media_type: Optional MIME type of the part content (e.g., "text/plain", "application/json", "image/png").

Returns:
A Part with the data field set.
"""
return Part(
data=ParseDict(data, struct_pb2.Value()),
media_type=media_type or '',
)


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