a2a.client.client module¶

class a2a.client.client.Client(interceptors: list[ClientCallInterceptor] | None = None)¶

Bases: ABC

Abstract base class defining the interface for an A2A client.

This class provides a standard set of methods for interacting with an A2A agent, regardless of the underlying transport protocol (e.g., gRPC, JSON-RPC). It supports sending messages, managing tasks, and handling event streams.

async add_interceptor(interceptor: ClientCallInterceptor) → None¶: Attaches additional interceptors to the Client.

abstractmethod async cancel_task(request: CancelTaskRequest, *, context: ClientCallContext | None = None) → Task¶: Requests the agent to cancel a specific task.

abstractmethod async close() → None¶: Closes the client and releases any underlying resources.

abstractmethod async create_task_push_notification_config(request: TaskPushNotificationConfig, *, context: ClientCallContext | None = None) → TaskPushNotificationConfig¶: Sets or updates the push notification configuration for a specific task.

abstractmethod async delete_task_push_notification_config(request: DeleteTaskPushNotificationConfigRequest, *, context: ClientCallContext | None = None) → None¶: Deletes the push notification configuration for a specific task.

abstractmethod async get_extended_agent_card(request: GetExtendedAgentCardRequest, *, context: ClientCallContext | None = None, signature_verifier: Callable[[AgentCard], None] | None = None) → AgentCard¶: Retrieves the agent’s card.

abstractmethod async get_task(request: GetTaskRequest, *, context: ClientCallContext | None = None) → Task¶: Retrieves the current state and history of a specific task.

abstractmethod async get_task_push_notification_config(request: GetTaskPushNotificationConfigRequest, *, context: ClientCallContext | None = None) → TaskPushNotificationConfig¶: Retrieves the push notification configuration for a specific task.

abstractmethod async list_task_push_notification_configs(request: ListTaskPushNotificationConfigsRequest, *, context: ClientCallContext | None = None) → ListTaskPushNotificationConfigsResponse¶: Lists push notification configurations for a specific task.

abstractmethod async list_tasks(request: ListTasksRequest, *, context: ClientCallContext | None = None) → ListTasksResponse¶: Retrieves tasks for an agent.

abstractmethod async send_message(request: SendMessageRequest, *, context: ClientCallContext | None = None) → AsyncIterator[StreamResponse]¶

Sends a message to the server.

This will automatically use the streaming or non-streaming approach as supported by the server and the client config. Client will aggregate update events and return an iterator of StreamResponse.

abstractmethod async subscribe(request: SubscribeToTaskRequest, *, context: ClientCallContext | None = None) → AsyncIterator[StreamResponse]¶: Resubscribes to a task’s event stream.

class a2a.client.client.ClientCallContext(*, state: MutableMapping[str, ~typing.Any]=<factory>, timeout: float | None = None, service_parameters: dict[str, str] | None=None)¶

Bases: BaseModel

A context passed with each client call, allowing for call-specific.

configuration and data passing. Such as authentication details or request deadlines.

model_config = {}¶: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

service_parameters: dict[str, str] | None¶

state: MutableMapping[str, Any]¶

timeout: float | None¶

class a2a.client.client.ClientConfig(streaming: bool = True, polling: bool = False, httpx_client: ~httpx.AsyncClient | None = None, grpc_channel_factory: ~collections.abc.Callable[[str], ~grpc.aio._base_channel.Channel] | None = None, supported_protocol_bindings: list[str] = <factory>, use_client_preference: bool = False, accepted_output_modes: list[str] = <factory>, push_notification_config: ~a2a_pb2.TaskPushNotificationConfig | None = None)¶

Bases: object

Configuration class for the A2AClient Factory.

accepted_output_modes: list[str]¶: The set of accepted output modes for the client.

grpc_channel_factory: Callable[[str], Channel] | None = None¶: Generates a grpc connection channel for a given url.

httpx_client: AsyncClient | None = None¶: Http client to use to connect to agent.

polling: bool = False¶

send. It is the callers job to check if the response is completed and if not run a polling loop.

Type:: Whether client prefers to poll for updates from message

push_notification_config: TaskPushNotificationConfig | None = None¶: Push notification configuration to use for every request.

streaming: bool = True¶: Whether client supports streaming

supported_protocol_bindings: list[str]¶

Ordered list of transports for connecting to agent (in order of preference). Empty implies JSONRPC only.

This is a string type to allow custom transports to exist in closed ecosystems.

use_client_preference: bool = False¶: Whether to use client transport preferences over server preferences. Recommended to use server preferences in most situations.