Google adk agent

Concrete implementation of AgentInterface using Google's Agent Development Kit (ADK).

This implementation wraps the official Google ADK Agent while maintaining compatibility with the AgentInterface. It leverages the async capabilities of ADK for optimal performance.

Authors

Raymond Christopher (raymond.christopher@gdplabs.id) Christian Trisno Sen Long Chen (christian.t.s.l.chen@gdplabs.id) Fachriza Dian Adhiatma (fachriza.d.adhiatma@gdplabs.id)

References

https://google.github.io/adk-docs/tools/mcp-tools/

A2ATool

Bases: BaseTool

A tool that communicates with an A2A agent.

A2AToolInput

Bases: BaseModel

Input for the A2ATool.

GoogleADKAgent(name, instruction, model, tools=None, description=None, max_iterations=3, agents=None, **kwargs)

Bases: BaseAgent

An agent that wraps a native Google ADK Agent.

This class implements the AgentInterface and uses Google's LlmAgent to handle the core conversation and tool execution logic via ADK's async-first design.

Initializes the GoogleADKAgent.

Parameters:

Name	Type	Description	Default
name	str	The name of this wrapper agent.	required
instruction	str	The instruction for this wrapper agent.	required
model	str	The name of the Google ADK model to use (e.g., "gemini-1.5-pro-latest").	required
tools	Optional[list[Any]]	An optional list of callable tools for the ADK agent.	None
description	Optional[str]	An optional human-readable description.	None
max_iterations	int	Maximum number of iterations to run (default: 3).	3
agents	Optional[List[GoogleADKAgent]]	Optional list of sub-agents that this agent can delegate to using ADK's built-in multi-agent capabilities. These will be passed as sub_agents to the underlying LlmAgent.	None
**kwargs	Any	Additional keyword arguments passed to the parent __init__.	{}

add_mcp_server(mcp_config)

Adds a new MCP server configuration.

Parameters:

Name	Type	Description	Default
mcp_config	Dict[str, Dict[str, Any]]	Dictionary containing server name as key and its configuration as value.	required

Raises:

Type	Description
ValueError	If mcp_config is empty or None, or if any server configuration is invalid.
KeyError	If any server name already exists in the configuration.

arun(query, **kwargs) async

Asynchronously runs the agent with the given query and returns the response.

Parameters:

Name	Type	Description	Default
query	str	The user's query to process.	required
**kwargs	Any	Additional keyword arguments. Supports "session_id", "user_id", "app_name".	{}

Returns:

Type	Description
Dict[str, Any]	A dictionary containing the output and other metadata.

arun_a2a_stream(query, configurable=None, **kwargs) async

Asynchronously streams the agent's response in a format compatible with A2A.

This method formats the ADK agent's streaming responses into a consistent format that the A2A executor can understand and process.

Parameters:

Name	Type	Description	Default
query	str	The input query for the agent.	required
configurable	Optional[Dict[str, Any]]	Optional dictionary for configuration, may include: - thread_id: The A2A task ID (used as session_id).	None
**kwargs	Any	Additional keyword arguments. Supports "user_id", "app_name".	{}

Yields:

Type	Description
AsyncGenerator[Dict[str, Any], None]	Dictionary with 'status' and 'content' fields that describe the agent's response state.

arun_stream(query, **kwargs) async

Runs the agent with the given query and streams the response parts.

Parameters:

Name	Type	Description	Default
query	str	The user's query to process.	required
**kwargs	Any	Additional keyword arguments. Supports "session_id", "user_id", "app_name".	{}

Yields:

Type	Description
AsyncIterator[str]	Text response chunks from the model. If an error occurs, the error message is yielded.

register_a2a_agents(agent_cards)

Convert known A2A agents to LangChain tools.

This method takes the agents from a2a_config.known_agents, creates A2AAgent instances for each one, and wraps them in LangChain tools.

Returns:

Name	Type	Description
None	None	The tools are added to the existing tools list.

run(query, **kwargs)

Synchronously runs the Google ADK agent by wrapping the internal async run method.

Parameters:

Name	Type	Description	Default
query	str	The input query for the agent.	required
**kwargs	Any	Additional keyword arguments passed to the internal async run method. Supports "session_id", "user_id", "app_name".	{}

Returns:

Type	Description
Dict[str, Any]	A dictionary containing the agent's response.

Raises:

Type	Description
RuntimeError	If asyncio.run() is called from an already running event loop, or for other unhandled errors during synchronous execution.

create_a2a_tool(agent_card)

Create a LangChain tool from an A2A agent card.