Base agent

Base class for concrete agent implementations.

This class provides common functionalities like A2A client capabilities.

Authors

Christian Trisno Sen Long Chen (christian.t.s.l.chen@gdplabs.id).

BaseAgent(name, instruction, description=None, model=None, tools=None, config=None, **kwargs)

Bases: AgentInterface

Base class for agents, providing common A2A client method implementations.

Concrete agent implementations (e.g., LangGraphAgent, GoogleADKAgent) should inherit from this class if they need to utilize the shared A2A client functionalities.

This class now supports flexible model handling: - model: Optional[Any] - can be an lm_invoker, string/ModelId, LangChain BaseChatModel, or other types - Automatically sets self.lm_invoker if an lm_invoker is provided or can be built - Stores the original model in self.model for subclass use

Initializes the BaseAgent.

Parameters:

Name	Type	Description	Default
name	str	The name of the agent.	required
instruction	str	The core directive or system prompt for the agent.	required
description	str | None	Human-readable description. Defaults to instruction if not provided.	None
model	Any | None	The model to use. Can be: - BaseLMInvoker instance (will be set as self.lm_invoker) - String or ModelId (will build an lm_invoker) - LangChain BaseChatModel (will be stored in self.model) - Any other type (will be stored in self.model)	None
tools	list[Any] | None	List of tools available to the agent.	None
config	BaseAgentConfig | dict[str, Any] | None	Additional configuration for the agent. Can be a BaseAgentConfig instance or dict.	None
**kwargs	Any	Additional keyword arguments for AgentInterface.	{}

model_provider: str property

Get the provider of the model with simplified logic.

Returns:

Name	Type	Description
str	str	The provider of the model.

add_mcp_server(mcp_config)

Adds MCP servers to the agent.

Parameters:

Name	Type	Description	Default
mcp_config	dict[str, dict[str, Any]]	A dictionary containing MCP server configurations.	required

Raises:

Type	Description
ValueError	If the MCP configuration is empty or None.
KeyError	If a server with the same name already exists in the MCP configuration.

asend_to_agent(agent_card, message, **kwargs) async

Asynchronously sends a message to another agent using the A2A protocol.

This method handles the core A2A communication logic, creating and sending properly formatted A2A messages and processing the responses.

Parameters:

Name	Type	Description	Default
agent_card	AgentCard	The AgentCard instance containing the target agent's details including URL, authentication requirements, and capabilities.	required
message	str | dict[str, Any]	The message to send to the agent. Can be either a string for simple text messages or a dictionary for structured data.	required
**kwargs	Any	Additional keyword arguments.	{}

Returns:

Type

Description

dict[str, Any]

A dictionary containing the response details: - status (str): 'success' or 'error' - content (str): Extracted text content from the response - task_id (str, optional): ID of the created/updated task - task_state (str, optional): Current state of the task - raw_response (str): Complete JSON response from the A2A client - error_type (str, optional): Type of error if status is 'error' - message (str, optional): Error message if status is 'error'

Raises:

Type	Description
HTTPError	If there's an HTTP-related error during the request.
Exception	For any other unexpected errors during message sending or processing.

astream_to_agent(agent_card, message, **kwargs) async

Asynchronously sends a streaming message to another agent using the A2A protocol.

This method supports streaming responses from the target agent, yielding chunks of the response as they become available. It handles various types of streaming events including task status updates, artifact updates, and message parts.

Parameters:

Name	Type	Description	Default
agent_card	AgentCard	The AgentCard instance containing the target agent's details including URL, authentication requirements, and capabilities.	required
message	str | dict[str, Any]	The message to send to the agent. Can be either a string for simple text messages or a dictionary for structured data.	required
**kwargs	Any	Additional keyword arguments.	{}

Yields:

Type

Description

AsyncGenerator[dict[str, Any], None]

Dictionaries containing streaming response chunks: For successful chunks: - status (str): 'success' - content (str): Extracted text content from the chunk - task_id (str): ID of the associated task - task_state (str): Current state of the task - final (bool): Whether this is the final chunk - artifact_name (str, optional): Name of the artifact if chunk is an artifact update For error chunks: - status (str): 'error' - error_type (str): Type of error encountered - message (str): Error description

Raises:

Type	Description
HTTPError	If there's an HTTP-related error during the streaming request.
Exception	For any other unexpected errors during message streaming or processing.

discover_agents(a2a_config, **kwargs) classmethod

Discover agents from the URLs specified in a2a_config.discovery_urls.

This concrete implementation fetches and parses .well-known/agent.json from each discovery URL to build a list of available agents.

Parameters:

Name	Type	Description	Default
a2a_config	A2AClientConfig	Configuration containing discovery URLs and other A2A settings.	required
**kwargs	Any	Additional keyword arguments (unused in this implementation).	{}

Returns:

Type	Description
list[AgentCard]	A list of AgentCard objects representing discovered agents.

format_agent_description(agent_card) staticmethod

Format the description of an agent card including skills information.

Parameters:

Name	Type	Description	Default
agent_card	AgentCard	The agent card to format.	required

Returns:

Name	Type	Description
str	str	The formatted description including skills.

get_name_preprocessor()

Get the name preprocessor based on the provider.

This will be used to correct the agent name and tool name. (mostly tool name)

Returns:

Name	Type	Description
NamePreprocessor	NamePreprocessor	The name preprocessor for the model.

send_to_agent(agent_card, message, **kwargs)

Synchronously sends a message to another agent using the A2A protocol.

This method is a synchronous wrapper around asend_to_agent. It handles the creation of an event loop if one doesn't exist, and manages the asynchronous call internally.

Parameters:

Name	Type	Description	Default
agent_card	AgentCard	The AgentCard instance containing the target agent's details including URL, authentication requirements, and capabilities.	required
message	str | dict[str, Any]	The message to send to the agent. Can be either a string for simple text messages or a dictionary for structured data.	required
**kwargs	Any	Additional keyword arguments passed to asend_to_agent.	{}

Returns:

Type

Description

dict[str, Any]

A dictionary containing the response details: - status (str): 'success' or 'error' - content (str): Extracted text content from the response - task_id (str, optional): ID of the created/updated task - task_state (str, optional): Current state of the task - raw_response (str): Complete JSON response from the A2A client - error_type (str, optional): Type of error if status is 'error' - message (str, optional): Error message if status is 'error'

Raises:

Type	Description
RuntimeError	If called from within an existing event loop or if asend_to_agent encounters an unhandled exception.

to_a2a(agent_card, **kwargs)

Converts the agent to an A2A-compatible ASGI application.

This implementation provides a base setup for A2A server components. Subclasses can override this method if they need custom executor or task store implementations.

Parameters:

Name	Type	Description	Default
agent_card	AgentCard	The agent card to use for the A2A application.	required
**kwargs	Any	Additional keyword arguments for ASGI application configuration.	{}

Returns:

Type	Description
Starlette	A Starlette ASGI application that can be used with any ASGI server.