a2anet 0.1.2

A package that makes implementing the A2A protocol easy

A2A Net

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.

📚 Table of Contents

• Installation
• Quick Start
• License
• Join the A2A Net Community

🛠️ Installation

To install with pip:

pip install a2anet

To install with uv:

uv add a2anet

🚀 Quick Start

Before 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.

Install LangGraph, tools, and A2A Net

First, LangGraph, the LangGraph Tavily API tool, and A2A Net.

To install with pip:

pip install langgraph langchain_tavily a2anet

To install with uv:

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:

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"
    "You should only respond to messages that can be answered by searching the web, and if the user's most recent message doesn't contain a question, or contains a question that can't be answered by searching the web, you should explain that to the user and ask them to try again with an appropriate query.\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"
    "The state of the task should be:\n"
    "- 'completed' if the user's most recent message contains a question that can be answered by searching the web, the `tavily_search` tool has been called, and the results are sufficient to answer the user's question.\n"
    "- 'failed' if the user's most recent message contains a question that can be answered by searching the web, the `tavily_search` tool has been called, and the results are insufficient to answer the user's question.\n"
    "- 'rejected' if the user's most recent message doesn't contain a question or contains a question that can't be answered by searching the web.\n"
    "If the task is 'completed', set 'task_state' to 'completed' and include at least one artifact in 'artifacts'.\n"
    "If the task is not 'completed', do not include any artifacts."
)

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 typing import List, Literal, Optional

from a2a.types import DataPart, TextPart
from pydantic import BaseModel, Field, model_validator


class Artifact(BaseModel):
    name: Optional[str] = Field(
        default=None,
        description="3-5 words describing the task output.",
    )
    description: Optional[str] = Field(
        default=None,
        description="1 sentence describing the task output.",
    )
    part: Optional[TextPart | DataPart] = Field(
        default=None,
        description="Task output. This can be a string, a markdown string, or a dictionary.",
    )


# The `TaskState`s are:
#
# submitted = 'submitted'
# working = 'working'
# input_required = 'input-required'
# completed = 'completed'
# canceled = 'canceled'
# failed = 'failed'
# rejected = 'rejected'
# auth_required = 'auth-required'
# unknown = 'unknown'
#
# `submitted`, `working`, `canceled`, and `unknown` are not decidable by the agent (they are handled in the `AgentExecutor`)
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"
        )
    )
    artifacts: Optional[List[Artifact]] = Field(
        default=None,
        description="Required if `task_state` is 'completed'. If `task_state` is not 'completed', `artifacts` should not be provided.",
    )

    @model_validator(mode="after")
    def _require_artifacts_when_completed(self):
        if self.task_state != "completed" and self.artifacts and len(self.artifacts) > 0:
            raise ValueError("`task_state` is not 'completed', `artifacts` should not be provided.")

        if self.task_state == "completed" and not (self.artifacts and len(self.artifacts) > 0):
            raise ValueError(
                "`task_state` is 'completed', `artifacts` must contain at least one item."
            )

        return self

Define the Agent's Agent Card

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?"],
    )],
)

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.

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.py

To run with uv:

uv run main.py

The server will start on http://localhost:8080.

📄 License

a2anet is distributed under the terms of the Apache-2.0 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