a2a.utils package¶

Submodules¶

a2a.utils.artifact module¶

Utility functions for creating A2A Artifact objects.

a2a.utils.artifact.new_artifact(parts: list[Part], name: str, description: str = '') → Artifact¶

Creates a new Artifact object.

Parameters:

parts – The list of Part objects forming the artifact’s content.
name – The human-readable name of the artifact.
description – An optional description of the artifact.

Returns:

A new Artifact object with a generated artifact_id.

a2a.utils.artifact.new_data_artifact(name: str, data: dict[str, Any], description: str = '') → Artifact¶

Creates a new Artifact object containing only a single DataPart.

Parameters:

name – The human-readable name of the artifact.
data – The structured data content of the artifact.
description – An optional description of the artifact.

Returns:

A new Artifact object with a generated artifact_id.

a2a.utils.artifact.new_text_artifact(name: str, text: str, description: str = '') → Artifact¶

Creates a new Artifact object containing only a single TextPart.

Parameters:

name – The human-readable name of the artifact.
text – The text content of the artifact.
description – An optional description of the artifact.

Returns:

A new Artifact object with a generated artifact_id.

a2a.utils.constants module¶

Constants for well-known URIs used throughout the A2A Python SDK.

a2a.utils.error_handlers module¶

a2a.utils.error_handlers.rest_error_handler(func: Callable[[...], Awaitable[Response]]) → Callable[[...], Awaitable[Response]]¶: Decorator to catch ServerError and map it to an appropriate JSONResponse.

a2a.utils.error_handlers.rest_stream_error_handler(func: Callable[[...], Coroutine[Any, Any, Any]]) → Callable[[...], Coroutine[Any, Any, Any]]¶: Decorator to catch ServerError for a straming method,log it and then rethrow it to be handled by framework.

a2a.utils.errors module¶

Custom exceptions for A2A server-side errors.

exception a2a.utils.errors.A2AServerError¶

Bases: Exception

Base exception for A2A Server errors.

exception a2a.utils.errors.MethodNotImplementedError(message: str = 'This method is not implemented by the server')¶

Bases: A2AServerError

Exception raised for methods that are not implemented by the server handler.

Bases: Exception

Wrapper exception for A2A or JSON-RPC errors originating from the server’s logic.

This exception is used internally by request handlers and other server components to signal a specific error that should be formatted as a JSON-RPC error response.

a2a.utils.helpers module¶

General utility functions for the A2A Python SDK.

a2a.utils.helpers.append_artifact_to_task(task: Task, event: TaskArtifactUpdateEvent) → None¶

Helper method for updating a Task object with new artifact data from an event.

Handles creating the artifacts list if it doesn’t exist, adding new artifacts, and appending parts to existing artifacts based on the append flag in the event.

Parameters:

task – The Task object to modify.
event – The TaskArtifactUpdateEvent containing the artifact data.

a2a.utils.helpers.are_modalities_compatible(server_output_modes: list[str] | None, client_output_modes: list[str] | None) → bool¶

Checks if server and client output modalities (MIME types) are compatible.

Modalities are compatible if: 1. The client specifies no preferred output modes (client_output_modes is None or empty). 2. The server specifies no supported output modes (server_output_modes is None or empty). 3. There is at least one common modality between the server’s supported list and the client’s preferred list.

Parameters:

server_output_modes – A list of MIME types supported by the server/agent for output. Can be None or empty if the server doesn’t specify.
client_output_modes – A list of MIME types preferred by the client for output. Can be None or empty if the client accepts any.

Returns:

True if the modalities are compatible, False otherwise.

a2a.utils.helpers.build_text_artifact(text: str, artifact_id: str) → Artifact¶

Helper to create a text artifact.

Parameters:

text – The text content for the artifact.
artifact_id – The ID for the artifact.

Returns:

An Artifact object containing a single TextPart.

a2a.utils.helpers.create_task_obj(message_send_params: MessageSendParams) → Task¶

Create a new task object from message send params.

Generates UUIDs for task and context IDs if they are not already present in the message.

Parameters:: message_send_params – The MessageSendParams object containing the initial message.
Returns:: A new Task object initialized with ‘submitted’ status and the input message in history.

a2a.utils.helpers.validate(expression: Callable[[Any], bool], error_message: str | None = None) → Callable¶

Decorator that validates if a given expression evaluates to True.

Typically used on class methods to check capabilities or configuration before executing the method’s logic. If the expression is False, a ServerError with an UnsupportedOperationError is raised.

Parameters:

expression – A callable that takes the instance (self) as its argument and returns a boolean.
error_message – An optional custom error message for the UnsupportedOperationError. If None, the string representation of the expression will be used.

a2a.utils.helpers.validate_async_generator(expression: Callable[[Any], bool], error_message: str | None = None)¶

Decorator that validates if a given expression evaluates to True.

Typically used on class methods to check capabilities or configuration before executing the method’s logic. If the expression is False, a ServerError with an UnsupportedOperationError is raised.

Parameters:

expression – A callable that takes the instance (self) as its argument and returns a boolean.
error_message – An optional custom error message for the UnsupportedOperationError. If None, the string representation of the expression will be used.

a2a.utils.message module¶

Utility functions for creating and handling A2A Message objects.

a2a.utils.message.get_data_parts(parts: list[Part]) → list[dict[str, Any]]¶

Extracts dictionary data from all DataPart objects in a list of Parts.

Parameters:: parts – A list of Part objects.
Returns:: A list of dictionaries containing the data from any DataPart objects found.

a2a.utils.message.get_file_parts(parts: list[Part]) → list[FileWithBytes | FileWithUri]¶

Extracts file data from all FilePart objects in a list of Parts.

Parameters:: parts – A list of Part objects.
Returns:: A list of FileWithBytes or FileWithUri objects containing the file data from any FilePart objects found.

a2a.utils.message.get_message_text(message: Message, delimiter: str = '\n') → str¶

Extracts and joins all text content from a Message’s parts.

Parameters:

message – The Message object.
delimiter – The string to use when joining text from multiple TextParts.

Returns:

A single string containing all text content, or an empty string if no text parts are found.

a2a.utils.message.get_text_parts(parts: list[Part]) → list[str]¶

Extracts text content from all TextPart objects in a list of Parts.

Parameters:: parts – A list of Part objects.
Returns:: A list of strings containing the text content from any TextPart objects found.

a2a.utils.message.new_agent_parts_message(parts: list[Part], context_id: str | None = None, task_id: str | None = None) → Message¶

Creates a new agent message containing a list of Parts.

Parameters:

parts – The list of Part objects for the message content.
context_id – The context ID for the message.
task_id – The task ID for the message.

Returns:

A new Message object with role ‘agent’.

a2a.utils.message.new_agent_text_message(text: str, context_id: str | None = None, task_id: str | None = None) → Message¶

Creates a new agent message containing a single TextPart.

Parameters:

text – The text content of the message.
context_id – The context ID for the message.
task_id – The task ID for the message.

Returns:

A new Message object with role ‘agent’.

a2a.utils.proto_utils module¶

Utils for converting between proto and Python types.

class a2a.utils.proto_utils.FromProto¶

Bases: object

Converts proto types to Python types.

classmethod agent_card(card: AgentCard) → AgentCard¶

classmethod agent_interface(interface: AgentInterface) → AgentInterface¶

classmethod artifact(artifact: Artifact) → Artifact¶

classmethod authentication_info(info: AuthenticationInfo) → PushNotificationAuthenticationInfo¶

classmethod capabilities(capabilities: AgentCapabilities) → AgentCapabilities¶

classmethod data(data: DataPart) → dict[str, Any]¶

classmethod file(file: FilePart) → FileWithUri | FileWithBytes¶

classmethod message(message: Message) → Message¶

classmethod message_send_configuration(config: SendMessageConfiguration) → MessageSendConfiguration¶

classmethod message_send_params(request: SendMessageRequest) → MessageSendParams¶

classmethod metadata(metadata: Struct) → dict[str, Any]¶

classmethod oauth2_flows(flows: OAuthFlows) → OAuthFlows¶

classmethod part(part: Part) → Part¶

classmethod provider(provider: AgentProvider | None) → AgentProvider | None¶

classmethod push_notification_config(config: PushNotificationConfig) → PushNotificationConfig¶

classmethod role(role: <google.protobuf.internal.enum_type_wrapper.EnumTypeWrapper object at 0x7fbd12d3cd10>) → Role¶

classmethod security(security: list[Security] | None) → list[dict[str, list[str]]] | None¶

classmethod security_scheme(scheme: SecurityScheme) → SecurityScheme¶

classmethod security_schemes(schemes: dict[str, SecurityScheme]) → dict[str, SecurityScheme]¶

classmethod skill(skill: AgentSkill) → AgentSkill¶

classmethod stream_response(response: StreamResponse) → Message | Task | TaskStatusUpdateEvent | TaskArtifactUpdateEvent¶

classmethod task(task: Task) → Task¶

classmethod task_artifact_update_event(event: TaskArtifactUpdateEvent) → TaskArtifactUpdateEvent¶

classmethod task_id_params(request: CancelTaskRequest | TaskSubscriptionRequest | GetTaskPushNotificationConfigRequest) → TaskIdParams¶

classmethod task_or_message(event: SendMessageResponse) → Task | Message¶

classmethod task_push_notification_config(config: TaskPushNotificationConfig) → TaskPushNotificationConfig¶

classmethod task_push_notification_config_request(request: CreateTaskPushNotificationConfigRequest) → TaskPushNotificationConfig¶

classmethod task_query_params(request: GetTaskRequest) → TaskQueryParams¶

classmethod task_state(state: <google.protobuf.internal.enum_type_wrapper.EnumTypeWrapper object at 0x7fbd131c88d0>) → TaskState¶

classmethod task_status(status: TaskStatus) → TaskStatus¶

classmethod task_status_update_event(event: TaskStatusUpdateEvent) → TaskStatusUpdateEvent¶

class a2a.utils.proto_utils.ToProto¶

Bases: object

Converts Python types to proto types.

classmethod agent_card(card: AgentCard) → AgentCard¶

classmethod agent_interface(interface: AgentInterface) → AgentInterface¶

classmethod artifact(artifact: Artifact) → Artifact¶

classmethod authentication_info(info: PushNotificationAuthenticationInfo) → AuthenticationInfo¶

classmethod capabilities(capabilities: AgentCapabilities) → AgentCapabilities¶

classmethod data(data: dict[str, Any]) → DataPart¶

classmethod file(file: FileWithUri | FileWithBytes) → FilePart¶

classmethod message(message: Message | None) → Message | None¶

classmethod message_send_configuration(config: MessageSendConfiguration | None) → SendMessageConfiguration¶

classmethod metadata(metadata: dict[str, Any] | None) → Struct | None¶

classmethod oauth2_flows(flows: OAuthFlows) → OAuthFlows¶

classmethod part(part: Part) → Part¶

classmethod provider(provider: AgentProvider | None) → AgentProvider | None¶

classmethod push_notification_config(config: PushNotificationConfig) → PushNotificationConfig¶

classmethod role(role: Role) → <google.protobuf.internal.enum_type_wrapper.EnumTypeWrapper object at 0x7fbd12d3cd10>¶

classmethod security(security: list[dict[str, list[str]]] | None) → list[Security] | None¶

classmethod security_scheme(scheme: SecurityScheme) → SecurityScheme¶

classmethod security_schemes(schemes: dict[str, SecurityScheme] | None) → dict[str, SecurityScheme] | None¶

classmethod skill(skill: AgentSkill) → AgentSkill¶

classmethod stream_response(event: Message | Task | TaskStatusUpdateEvent | TaskArtifactUpdateEvent) → StreamResponse¶

classmethod task(task: Task) → Task¶

classmethod task_artifact_update_event(event: TaskArtifactUpdateEvent) → TaskArtifactUpdateEvent¶

classmethod task_or_message(event: Task | Message) → SendMessageResponse¶

classmethod task_push_notification_config(config: TaskPushNotificationConfig) → TaskPushNotificationConfig¶

classmethod task_state(state: TaskState) → <google.protobuf.internal.enum_type_wrapper.EnumTypeWrapper object at 0x7fbd131c88d0>¶

classmethod task_status(status: TaskStatus) → TaskStatus¶

classmethod task_status_update_event(event: TaskStatusUpdateEvent) → TaskStatusUpdateEvent¶

classmethod update_event(event: Task | Message | TaskStatusUpdateEvent | TaskArtifactUpdateEvent) → StreamResponse¶: Converts a task, message, or task update event to a StreamResponse.

a2a.utils.task module¶

Utility functions for creating A2A Task objects.

a2a.utils.task.completed_task(task_id: str, context_id: str, artifacts: list[Artifact], history: list[Message] | None = None) → Task¶

Creates a Task object in the ‘completed’ state.

Useful for constructing a final Task representation when the agent finishes and produces artifacts.

Parameters:

task_id – The ID of the task.
context_id – The context ID of the task.
artifacts – A list of Artifact objects produced by the task.
history – An optional list of Message objects representing the task history.

Returns:

A Task object with status set to ‘completed’.

a2a.utils.task.new_task(request: Message) → Task¶

Creates a new Task object from an initial user message.

Generates task and context IDs if not provided in the message.

Parameters:

request – The initial Message object from the user.

Returns:

A new Task object initialized with ‘submitted’ status and the input message in history.

Raises:

TypeError – If the message role is None.
ValueError – If the message parts are empty, if any part has empty content, or if the provided context_id is invalid.

a2a.utils.telemetry module¶

OpenTelemetry Tracing Utilities for A2A Python SDK.

This module provides decorators to simplify the integration of OpenTelemetry tracing into Python applications. It offers trace_function for instrumenting individual functions (both synchronous and asynchronous) and trace_class for instrumenting multiple methods within a class.

The tracer is initialized with the module name and version defined by INSTRUMENTING_MODULE_NAME (‘a2a-python-sdk’) and INSTRUMENTING_MODULE_VERSION (‘1.0.0’).

Features: - Automatic span creation for decorated functions/methods. - Support for both synchronous and asynchronous functions. - Default span naming based on module and function/class/method name. - Customizable span names, kinds, and static attributes. - Dynamic attribute setting via an attribute_extractor callback. - Automatic recording of exceptions and setting of span status. - Selective method tracing in classes using include/exclude lists.

Usage:

For a single function: ```python from your_module import trace_function

@trace_function def my_function():

# … pass

@trace_function(span_name=’custom.op’, kind=SpanKind.CLIENT) async def my_async_function():

# … pass

```

For a class: ```python from your_module import trace_class

@trace_class(exclude_list=[‘internal_method’]) class MyService:

def public_api(self, user_id):
# This method will be traced pass

def internal_method(self):
# This method will not be traced pass

```

Module contents¶

Utility functions for the A2A Python SDK.

a2a.utils.append_artifact_to_task(task: Task, event: TaskArtifactUpdateEvent) → None¶

Helper method for updating a Task object with new artifact data from an event.

Handles creating the artifacts list if it doesn’t exist, adding new artifacts, and appending parts to existing artifacts based on the append flag in the event.

Parameters:

task – The Task object to modify.
event – The TaskArtifactUpdateEvent containing the artifact data.

a2a.utils.are_modalities_compatible(server_output_modes: list[str] | None, client_output_modes: list[str] | None) → bool¶

Checks if server and client output modalities (MIME types) are compatible.

Parameters:

server_output_modes – A list of MIME types supported by the server/agent for output. Can be None or empty if the server doesn’t specify.
client_output_modes – A list of MIME types preferred by the client for output. Can be None or empty if the client accepts any.

Returns:

True if the modalities are compatible, False otherwise.

a2a.utils.build_text_artifact(text: str, artifact_id: str) → Artifact¶

Helper to create a text artifact.

Parameters:

text – The text content for the artifact.
artifact_id – The ID for the artifact.

Returns:

An Artifact object containing a single TextPart.

a2a.utils.completed_task(task_id: str, context_id: str, artifacts: list[Artifact], history: list[Message] | None = None) → Task¶

Creates a Task object in the ‘completed’ state.

Useful for constructing a final Task representation when the agent finishes and produces artifacts.

Parameters:

task_id – The ID of the task.
context_id – The context ID of the task.
artifacts – A list of Artifact objects produced by the task.
history – An optional list of Message objects representing the task history.

Returns:

A Task object with status set to ‘completed’.

a2a.utils.create_task_obj(message_send_params: MessageSendParams) → Task¶

Create a new task object from message send params.

Generates UUIDs for task and context IDs if they are not already present in the message.

Parameters:: message_send_params – The MessageSendParams object containing the initial message.
Returns:: A new Task object initialized with ‘submitted’ status and the input message in history.

a2a.utils.get_data_parts(parts: list[Part]) → list[dict[str, Any]]¶

Extracts dictionary data from all DataPart objects in a list of Parts.

Parameters:: parts – A list of Part objects.
Returns:: A list of dictionaries containing the data from any DataPart objects found.

a2a.utils.get_file_parts(parts: list[Part]) → list[FileWithBytes | FileWithUri]¶

Extracts file data from all FilePart objects in a list of Parts.

Parameters:: parts – A list of Part objects.
Returns:: A list of FileWithBytes or FileWithUri objects containing the file data from any FilePart objects found.

a2a.utils.get_message_text(message: Message, delimiter: str = '\n') → str¶

Extracts and joins all text content from a Message’s parts.

Parameters:

message – The Message object.
delimiter – The string to use when joining text from multiple TextParts.

Returns:

A single string containing all text content, or an empty string if no text parts are found.

a2a.utils.get_text_parts(parts: list[Part]) → list[str]¶

Extracts text content from all TextPart objects in a list of Parts.

Parameters:: parts – A list of Part objects.
Returns:: A list of strings containing the text content from any TextPart objects found.

a2a.utils.new_agent_parts_message(parts: list[Part], context_id: str | None = None, task_id: str | None = None) → Message¶

Creates a new agent message containing a list of Parts.

Parameters:

parts – The list of Part objects for the message content.
context_id – The context ID for the message.
task_id – The task ID for the message.

Returns:

A new Message object with role ‘agent’.

a2a.utils.new_agent_text_message(text: str, context_id: str | None = None, task_id: str | None = None) → Message¶

Creates a new agent message containing a single TextPart.

Parameters:

text – The text content of the message.
context_id – The context ID for the message.
task_id – The task ID for the message.

Returns:

A new Message object with role ‘agent’.

a2a.utils.new_artifact(parts: list[Part], name: str, description: str = '') → Artifact¶

Creates a new Artifact object.

Parameters:

parts – The list of Part objects forming the artifact’s content.
name – The human-readable name of the artifact.
description – An optional description of the artifact.

Returns:

A new Artifact object with a generated artifact_id.

a2a.utils.new_data_artifact(name: str, data: dict[str, Any], description: str = '') → Artifact¶

Creates a new Artifact object containing only a single DataPart.

Parameters:

name – The human-readable name of the artifact.
data – The structured data content of the artifact.
description – An optional description of the artifact.

Returns:

A new Artifact object with a generated artifact_id.

a2a.utils.new_task(request: Message) → Task¶

Creates a new Task object from an initial user message.

Generates task and context IDs if not provided in the message.

Parameters:

request – The initial Message object from the user.

Returns:

A new Task object initialized with ‘submitted’ status and the input message in history.

Raises:

TypeError – If the message role is None.
ValueError – If the message parts are empty, if any part has empty content, or if the provided context_id is invalid.

a2a.utils.new_text_artifact(name: str, text: str, description: str = '') → Artifact¶

Creates a new Artifact object containing only a single TextPart.

Parameters:

name – The human-readable name of the artifact.
text – The text content of the artifact.
description – An optional description of the artifact.

Returns:

A new Artifact object with a generated artifact_id.