nat.plugins.a2a.client.client_impl#

Attributes#

logger

Classes#

`GetTaskInput`	Input for get_task function.
`CancelTaskInput`	Input for cancel_task function.
`SendMessageInput`	Input for send_message function.
`A2AClientFunctionGroup`	A minimal FunctionGroup for A2A agents.

Functions#

a2a_client_function_group(config, _builder)

Connect to an A2A agent, discover agent card and publish the primary

Module Contents#

logger#

class GetTaskInput(/, **data: Any)#

Bases: pydantic.BaseModel

Input for get_task function.

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

task_id: str = None#

history_length: int | None = None#

class CancelTaskInput(/, **data: Any)#

Bases: pydantic.BaseModel

Input for cancel_task function.

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

task_id: str = None#

class SendMessageInput(/, **data: Any)#

Bases: pydantic.BaseModel

Input for send_message function.

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

query: str = None#

task_id: str | None = None#

context_id: str | None = None#

class A2AClientFunctionGroup( config: nat.plugins.a2a.client.client_config.A2AClientConfig, builder: nat.builder.workflow_builder.Builder, )#

Bases: nat.builder.function.FunctionGroup

A minimal FunctionGroup for A2A agents.

Exposes a simple send_message function to interact with A2A agents.

Creates a new function group.

Parameters#

configFunctionGroupBaseConfig: The configuration for the function group.
instance_namestr | None, optional: The name of the function group. If not provided, the type of the function group will be used.
filter_fnCallable[[Sequence[str]], Awaitable[Sequence[str]]] | None, optional: A callback function to additionally filter the functions in the function group dynamically when the functions are accessed via any accessor method.
middlewareSequence[Middleware] | None, optional: The middleware instances to apply to all functions in this group.

_builder#

_client: nat.plugins.a2a.client.client_base.A2ABaseClient | None = None#

_include_skills_in_description#

_register_functions()#: Retrieve agent card and register the three-level API: high-level, helpers, and low-level.

_format_main_function_description(agent_card) → str#: Create description for the main agent function.

_create_high_level_function()#: High-level function that simplifies the response.

async _get_skills(params: dict | None = None) → dict#: Helper function to list agent skills.

async _get_agent_info(params: dict | None = None) → dict#: Helper function to get agent metadata.

async _wrap_get_task(params: GetTaskInput) → Any#: Wrapper for get_task that delegates to client_base.

async _wrap_cancel_task(params: CancelTaskInput) → Any#: Wrapper for cancel_task that delegates to client_base.

async _send_message_advanced(params: SendMessageInput) → list#

Send a message with full A2A protocol control.

Returns: List of ClientEvent|Message objects containing: - Task information - Status updates - Artifact updates - Full message details

async _send_message_streaming( params: SendMessageInput, ) → collections.abc.AsyncGenerator[Any, None]#

Send a message with full A2A protocol control and stream events.

This is an async generator that yields events as they arrive from the agent.

Yields: ClientEvent|Message objects containing: - Task information - Status updates - Artifact updates - Full message details

async a2a_client_function_group( config: nat.plugins.a2a.client.client_config.A2AClientConfig, _builder: nat.builder.workflow_builder.Builder, )#

Connect to an A2A agent, discover agent card and publish the primary agent function and helper functions. This function group is per-user, meaning each user gets their own isolated instance.

This function group creates a three-level API: - High-level: Agent function named after the agent (e.g., dice_agent) - Helpers: get_skills, get_info, get_task, cancel_task - Low-level: send_message for advanced usage