Note
Go to the end to download the full example code.
A2A Agent¶
A2A (Agent-to-Agent) is an open standard protocol for enabling interoperable communication between different AI agents.
AgentScope provides support for the A2A protocol at two levels: obtaining Agent Card information and connecting to remote agents. The related APIs are as follows:
Class |
Description |
|---|---|
|
Agent class for communicating with remote A2A agents |
|
Formatter for converting between AgentScope messages and A2A message/task formats |
|
Base class for Agent Card resolvers |
|
Resolver for loading Agent Cards from local JSON files |
|
Resolver for fetching Agent Cards from the well-known path of a URL |
|
Resolver for fetching Agent Cards from the Nacos Agent Registry |
This section demonstrates how to create an A2AAgent and communicate with remote A2A agents.
Note
Note that A2A support is an experimental feature and may change in future versions. Due to limitations of the A2A protocol itself, A2AAgent cannot fully align with local agents like ReActAgent, including:
Only supports chatbot scenarios, i.e., only supports conversations between one user and one agent (does not affect handoff/router usage patterns)
Does not support real-time interruption during conversations
Does not support agentic structured output
In the current implementation, messages received by the
observemethod are stored locally and sent to the remote agent together when thereplymethod is called. Therefore, if severalobservecalls are made without a subsequentreplycall, those messages will not be seen by the remote agent
from a2a.types import AgentCard, AgentCapabilities
from v2.nacos import ClientConfig
from agentscope.a2a import WellKnownAgentCardResolver, NacosAgentCardResolver
from agentscope.agent import A2AAgent, UserAgent
from agentscope.message import Msg, TextBlock
from agentscope.tool import ToolResponse
/opt/hostedtoolcache/Python/3.10.19/x64/lib/python3.10/site-packages/v2/nacos/naming/model/service.py:20: PydanticDeprecatedSince20: Support for class-based `config` is deprecated, use ConfigDict instead. Deprecated in Pydantic V2.0 to be removed in V3.0. See Pydantic V2 Migration Guide at https://errors.pydantic.dev/2.12/migration/
class Service(BaseModel):
Obtaining Agent Cards¶
First, we need to obtain an Agent Card to connect to the corresponding agent. An Agent Card contains information such as the agent’s name, description, capabilities, and connection details.
Manually Creating Agent Card¶
If you know all the information of an Agent Card, you can directly create an Agent Card object from a2a.types.AgentCard.
# Create an Agent Card object
agent_card = AgentCard(
name="Friday", # Agent name
description="A fun chatting companion", # Agent description
url="http://localhost:8000", # Agent's RPC service URL
version="1.0.0", # Agent version
capabilities=AgentCapabilities( # Agent capability configuration
push_notifications=False,
state_transition_history=True,
streaming=True,
),
default_input_modes=["text/plain"], # Supported input formats
default_output_modes=["text/plain"], # Supported output formats
skills=[], # Agent skill list
)
Fetching from Remote Services¶
AgentScope also supports fetching from the standard path of remote services (well-known server).
Here’s an example using WellKnownAgentCardResolver to fetch an Agent Card from the standard path of a remote service:
async def agent_card_from_well_known_website() -> AgentCard:
"""Example of fetching an Agent Card from the well-known path of a remote service."""
# Create an Agent Card resolver
resolver = WellKnownAgentCardResolver(
base_url="http://localhost:8000",
)
# Fetch and return the Agent Card
return await resolver.get_agent_card()
Loading Agent Cards from Local Files¶
The FileAgentCardResolver class supports loading Agent Cards from local JSON files, suitable for configuration file management scenarios.
An example of an Agent Card in JSON format is shown below:
{
"name": "RemoteAgent",
"url": "http://localhost:8000",
"description": "Remote A2A Agent",
"version": "1.0.0",
"capabilities": {},
"default_input_modes": ["text/plain"],
"default_output_modes": ["text/plain"],
"skills": []
}
You can easily load this file using FileAgentCardResolver:
async def agent_card_from_file() -> AgentCard:
"""Example of loading an Agent Card from a local JSON file."""
from agentscope.a2a import FileAgentCardResolver
# Load Agent Card from JSON file
resolver = FileAgentCardResolver(
file_path="./agent_card.json", # JSON file path
)
# Fetch and return the Agent Card
return await resolver.get_agent_card()
Fetching Agent Cards from Nacos Registry¶
Nacos is an open-source dynamic service discovery, configuration management, and service management platform. In version 3.1.0, it introduced the Agent Registry feature, supporting distributed registration, discovery, and version management of A2A agents.
Important
The prerequisite for using NacosAgentCardResolver is that the user has deployed a Nacos server version 3.1.0 or higher. For deployment and registration procedures, please refer to the official documentation.
async def agent_card_from_nacos() -> AgentCard:
"""Example of fetching an Agent Card from the Nacos registry."""
# Create a Nacos Agent Card resolver
resolver = NacosAgentCardResolver(
remote_agent_name="my-remote-agent", # Agent name registered in Nacos
nacos_client_config=ClientConfig(
server_addresses="http://localhost:8848", # Nacos server address
# Other optional configuration items
),
)
# Fetch and return the Agent Card
return await resolver.get_agent_card()
Building an A2A Agent¶
The A2AAgent class provided by AgentScope is used to communicate with remote A2A agents, and its usage is similar to regular agents.
agent = A2AAgent(agent_card=agent_card)
Using A2AAgent, developers can build chatbot scenario conversations, or encapsulate it as a tool function to build more complex application scenarios such as handoff/router.
Currently, the format protocol conversion supported by A2AAgent is handled by agentscope.formatter.A2AChatFormatter, which supports:
Converting AgentScope’s
Msgmessages to A2A protocol’sMessageformatConverting A2A protocol responses back to AgentScope’s
MsgformatConverting A2A protocol’s
Taskresponses to AgentScope’sMsgformatSupporting multiple content types such as text, images, audio, and video
async def a2a_in_chatbot() -> None:
"""Example of chatting using A2AAgent."""
user = UserAgent("user")
msg = None
while True:
msg = await user(msg)
if msg.get_text_content() == "exit":
break
msg = await agent(msg)
Or encapsulate it as a tool function for invocation:
async def create_worker(query: str) -> ToolResponse:
"""Complete a given task through a sub-agent
Args:
query (`str`):
Description of the task to be completed by the sub-agent
"""
res = await agent(
Msg("user", query, "user"),
)
return ToolResponse(
content=[
TextBlock(
type="text",
text=res.get_text_content(),
),
],
)
Total running time of the script: (0 minutes 0.575 seconds)