Content
# A2A Net
[](https://pypi.org/project/a2anet)
[](https://pypi.org/project/a2anet)
A2A Net is a package for easy [A2A protocol](https://a2aproject.github.io/A2A/latest/) implementation.
A2A was designed for AI agent communication and collaboration, but many people use [MCP](https://modelcontextprotocol.io/introduction) for agent communication, despite the fact that MCP was designed for tools, and [agents are not tools](https://www.googlecloudcommunity.com/gc/Community-Blogs/Agents-are-not-tools/ba-p/922716)!
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](https://a2aproject.github.io/A2A/latest/topics/key-concepts/).
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](#installation) and [Quick Start](#quick-start) to get started.
## Table of Contents
- [Installation](#installation)
- [Quick Start](#quick-start)
- [License](#license)
- [Join the A2A Net Community](#join-the-a2a-net-community)
## Installation
To install with pip:
```console
pip install a2anet
```
To install with uv:
```console
uv add a2anet
```
## Quick Start
Before going through the Quick Start it might be helpful to read [Key Concepts in A2A](https://a2aproject.github.io/A2A/latest/topics/key-concepts/), especially the "Core Actors" and "Fundamental Communication Elements" sections.
For an example agent that uses A2A Net, see [Tavily Agent](https://github.com/A2ANet/TavilyAgent).
### Install LangGraph, tools, and A2A Net
First, LangGraph, the LangGraph Tavily API tool, and A2A Net.
To install with pip:
```console
pip install langgraph langchain_tavily a2anet
```
To install with uv:
```console
uv add langgraph langchain_tavily a2anet
```
### Create a ReAct Agent with LangGraph
Then, 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`:
```python
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 `Message`s whilst keeping the `Task`'s output concise for the receiving person or agent.
`src/a2anet/types/langgraph.py`:
```python
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.",
)
```
### Define the Agent's Agent Card
The Agent Card is an essential component of [agent discovery](https://a2aproject.github.io/A2A/v0.2.5/topics/agent-discovery/).
It allows people and other agents to browse agents and their skills.
```python
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?"],
)],
)
```
### Create the Agent Executor, Request Handler, and A2A Server
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`.
```python
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:
```python
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](https://github.com/A2ANet/A2AUI) or [A2A Protocol Inspector](https://github.com/a2aproject/a2a-inspector).
To run with python:
```console
python main.py
```
To run with uv:
```console
uv run main.py
```
The server will start on `http://localhost:8080`.
## License
`a2anet` is distributed under the terms of the [Apache-2.0](https://spdx.org/licenses/Apache-2.0.html) license.
## Join the A2A Net Community
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
