A2A Net is a package for easy A2A protocol implementation. A2A was designed for AI agent communication and collaboration, but many people use MCP for agent communication, despite the fact that MCP was designed for tools, and agents are not tools!
This is likely due to a number of reasons, e.g. MCP has been around for longer, tool use is more common than multi-agent systems, more people are familiar with MCP, etc. However, there is also a reason independent of MCP: A2A has a steep learning curve.
Agent communication and collaboration is more complicated than tool use, and A2A introduces a number of concepts like: A2A Client, A2A Server, Agent Card, Message, Task, Part, Artifact, and more. For example, an A2A Client is an application or agent that initiates requests to an A2A Server on behalf of a user or another system, and an Artifact is an output (e.g., a document, image, structured data) generated by the agent as a result of a Task, composed of Parts.
Implementing A2A requires learning about all of these concepts, and then creating at least three files that contain 100s of lines of code: main.py, agent.py, and agent_executor.py.
The aim of this package is to reduce the learning curve and encourage A2A use.
With A2A Net, it's possible to create an A2A agent with one main.py file in less than 100 lines of code.
It does this by defining an AgentExecutor object for each agent framework (e.g. LangGraph) which converts known framework objects (e.g. AIMessage) to A2A objects (e.g. Message).
AgentExecutor is fully customisable, methods like _handle_ai_message can be overridden to change its behaviour.
See Installation and Quick Start to get started.
To install with pip:
pip install a2anetTo install with uv:
uv add a2anetBefore going through the Quick Start it might be helpful to read Key Concepts in A2A, especially the "Core Actors" and "Fundamental Communication Elements" sections.
For an example agent that uses A2A Net, see Tavily Agent.
First, LangGraph, the LangGraph Tavily API tool, and A2A Net.
To install with pip:
pip install langgraph langchain_tavily a2anetTo install with uv:
uv add langgraph langchain_tavily a2anetThen, create a ReAct Agent with LangGraph.
The RESPONSE_FORMAT_INSTRUCTION and StructuredResponse are for the A2A protocol.
First, the user's query is processed with the SYSTEM_INSTRUCTION in a loop until the agent exits.
Once the agent has exited, an LLM is called to produce a StructuredResponse with the RESPONSE_FORMAT_INSTRUCTION, the user's query, and the agent's messages and tool calls.
main.py:
from a2anet.types.langgraph import StructuredResponse # For the A2A protocol
from langchain_tavily import TavilySearch
from langgraph.checkpoint.memory import MemorySaver
from langgraph.prebuilt import create_react_agent
SYSTEM_INSTRUCTION: str = (
"You are a helpful assistant that can search the web with the Tavily API and answer questions about the results.\n"
"If the `tavily_search` tool returns insufficient results, you should explain that to the user and ask them to try again with a more specific query.\n"
"You can use markdown format to format your responses."
)
# For the A2A protocol
RESPONSE_FORMAT_INSTRUCTION: str = (
"You are an expert A2A protocol agent.\n"
"Your task is to read through all previous messages thoroughly and determine what the state of the task is.\n"
"If the task is complete, extract the task output into an artifact. The task is complete if the `tavily_search` tool has been called and the results are sufficient to answer the user's question."
)
graph = create_react_agent(
model="anthropic:claude-sonnet-4-20250514",
tools=[TavilySearch(max_results=2)],
checkpointer=MemorySaver(),
prompt=SYSTEM_INSTRUCTION,
response_format=(RESPONSE_FORMAT_INSTRUCTION, StructuredResponse), # For the A2A protocol
)StructuredResponse is a Pydantic object that represents the Task's state, and if the task has been "completed", the Task's Artifact.
For example, if the user's query was "Hey!", the task_state should be "input-required", because the Tavily Agent's task is to call the tavily_search tool and answer the user's question.
If the tavily_search tool returns insufficient results, the task_state might be "failed".
On the other hand, if the tavily_search tool has been called and the results are sufficient to answer the user's question, the task_state should be "completed".
Task states like "completed", "input-required", and "failed" help people and agents keep track of a Task's progress, and whilst it's probably overkill for an interaction between a single person and agent, they become essential as the system grows in complexity.
A Task's output is an Artifact and is distinct from a Message.
This allows the agent to share its progress (and optionally, steps) with Messages whilst keeping the Task's output concise for the receiving person or agent.
src/a2anet/types/langgraph.py:
from pydantic import BaseModel
class StructuredResponse(BaseModel):
task_state: Literal[
"input-required",
"completed",
"failed",
"rejected",
"auth-required",
] = Field(
description=(
"The state of the task:\n"
"- 'input-required': The task requires additional input from the user.\n"
"- 'completed': The task has been completed.\n"
"- 'failed': The task has failed.\n"
"- 'rejected': The task has been rejected.\n"
"- 'auth-required': The task requires authentication from the user.\n"
)
)
task_state_message: str = Field(
description=(
"A message to the user about the state of the task. "
"If the state is 'input-required' or 'auth-required', it should explain what input or authentication is required to complete the task."
)
)
artifact_title: Optional[str] = Field(
default=None,
description="Required if the `task_state` is 'completed'. 3-5 words describing the task output.",
)
artifact_description: Optional[str] = Field(
default=None,
description="Required if the `task_state` is 'completed'. 1 sentence describing the task output.",
)
artifact_output: Optional[str] = Field(
default=None,
description="Required if the `task_state` is 'completed'. The task output. This can be a string, a markdown string, or a string that is parsable as JSON.",
)The Agent Card is an essential component of agent discovery. It allows people and other agents to browse agents and their skills.
from a2a.types import AgentCapabilities, AgentCard, AgentSkill
agent_card: AgentCard = AgentCard(
name="Tavily Agent",
description="Search the web with the Tavily API and answer questions about the results.",
url="http://localhost:8080",
version="1.0.0",
defaultInputModes=["text", "text/plain"],
defaultOutputModes=["text", "text/plain"],
capabilities=AgentCapabilities(),
skills=[AgentSkill(
id="search-web",
name="Search Web",
description="Search the web with the Tavily API and answer questions about the results.",
tags=["search", "web", "tavily"],
examples=["Who is Leo Messi?"],
)],
)This is where the magic happens... instead of creating agent.py and agent_executor.py files, simply pass the ReAct Agent graph we defined eariler to LangGraphAgentExecutor.
import uvicorn
from a2a.server.apps import A2AStarletteApplication
from a2a.server.request_handlers import DefaultRequestHandler
from a2a.server.tasks import InMemoryTaskStore
from a2anet.executors.langgraph import LangGraphAgentExecutor
agent_executor: LangGraphAgentExecutor = LangGraphAgentExecutor(graph)
request_handler: DefaultRequestHandler = DefaultRequestHandler(
agent_executor=agent_executor, task_store=InMemoryTaskStore()
)
server: A2AStarletteApplication = A2AStarletteApplication(
agent_card=agent_card, http_handler=request_handler
)
uvicorn.run(server.build(), host="0.0.0.0", port=port)If you want to change the behaviour of a method in LangGraphAgentExecutor you can override it!
For example:
from a2a.server.tasks.task_updater import TaskUpdater
from a2a.types import Task
from langchain_core.messages import AIMessage
class MyLangGraphAgentExecutor(LangGraphAgentExecutor):
async def _handle_ai_message(self, message: AIMessage, task: Task, task_updater: TaskUpdater):
print("Hello World!")
agent_executor: LangGraphAgentExecutor = MyLangGraphAgentExecutor(graph)That's it! Run main.py and test the agent with the Agent2Agent (A2A) UI or A2A Protocol Inspector.
To run with python:
python main.pyTo run with uv:
uv run main.pyThe server will start on http://localhost:8080.
a2anet is distributed under the terms of the Apache-2.0 license.
A2A Net is a site to find and share AI agents and open-source community. Join to share your A2A agents, ask questions, stay up-to-date with the latest A2A news, be the first to hear about open-source releases, tutorials, and more!
- 🌍 Site: https://a2anet.com/
- 🤖 Discord: https://discord.gg/674NGXpAjU