A2a Protocol

AI Agent

Python

Generative Ai Tools

Generative Ai Use Cases

The Ultimate A2A Handbook: Rulebook for Agent Conversations

As artificial intelligence continues to evolve, the need for AI systems to communicate and collaborate has become increasingly important. From document summarization to image generation and intelligent decision-making, today’s AI agents are no longer working in isolation — they must interact to handle complex tasks effectively. This is where the Agent-to-Agent (A2A) Protocol comes into play.

A2A is a communication framework designed to enable seamless interaction between autonomous agents. In an era dominated by distributed systems and multi-agent architectures, A2A offers a structured way for intelligent agents to share information, delegate tasks, and make coordinated decisions without human intervention.

This article explores the A2A protocol from foundational concepts to advanced implementations, and a building a travel planner app using A2A.

Getting Started

Table of contents

• What is A2A
• Core concepts of A2A
• A2A vs MCP
• Experimenting with A2A
• 1. Implementing A2A from scratch using FastAPI
• 2. Implementing A2A using Google A2A SDK
• A2A Client-Server Interaction Flow
• 3. Implementing A2A using Python-A2A library
• Example 1: Echo Agent
• Example 2: Basic A2A Agent
• Example 3: LLM-Based Agent
• Example 4: Converting LangChain to A2A Servers
• Example 5: Converting MCP Tools to LangChain Tools
• Building a travel planner app using A2A

What is A2A

Image Source: A Visual Guide to Agent2Agent (A2A) Protocol

Imagine assembling a team of exceptional AI assistants — one masters data analysis, another crafts insightful reports, and a third flawlessly manages your schedule. Individually, they’re outstanding. But there’s a hitch: each speaks a different language. One uses Python, another JSON, and the third relies on obscure API calls. Getting them to collaborate would be like reviving the digital Tower of Babel. This is the challenge that the Agent-to-Agent (A2A) Protocol is designed to solve.

The Agent-to-Agent (A2A) Protocol, introduced by Google Cloud, is an open standard that enables seamless communication and collaboration between AI agents — regardless of the frameworks or vendors they originate from. Like a universal translator, A2A solves the interoperability challenge by providing a common language for agents to share information, delegate tasks, and coordinate actions effectively.

A2A complements Anthropic’s Model Context Protocol (MCP) by focusing on inter-agent communication, leveraging Google’s expertise in deploying large-scale agent systems. Together, these protocols lay the foundation for the future of multi-agent collaboration in enterprise environments. The protocol is supported by over 50 major technology and consulting partners, reflecting a shared vision for scalable, interoperable agent ecosystems.

Core concepts of A2A

The Agent-to-Agent (A2A) Protocol is designed around a set of foundational concepts that enable intelligent agents to collaborate efficiently and reliably. These core building blocks define how agents communicate, manage tasks, and exchange data.

• AgentCard
  A standardized JSON document that describes an agent’s identity, capabilities, and supported protocols. Typically hosted at the /.well-known/agent.json endpoint, the AgentCard allows other agents or clients to easily discover and understand how to interact with the agent.
• Task
  A Task represents a stateful collaboration between a client and an agent, tracking progress toward a specific goal. It includes task status, execution history, and references to outputs (artifacts). Tasks enable agents to maintain context across multi-step processes.
• Artifact
  An Artifact is the final, immutable result produced by an agent during a task. It may include one or more Parts, which are discrete pieces of structured or unstructured content (e.g., text, files, or forms). Artifacts are useful for recording outcomes or sharing final deliverables.
• Message
  Used to exchange non-artifact content between agents or with clients. Messages may contain instructions, intermediate thoughts, context updates, or task status information. They support dynamic interaction throughout the lifecycle of a task.
• Part
  A Part is the smallest unit of content within a Message or Artifact. Each Part has a specific content type (e.g., text/plain, application/json, or a file reference), allowing agents to structure complex content with clarity and modularity.
• Transport mechanism
  The transport mechanism in the A2A protocol determines how agents exchange messages, using technologies like HTTP/HTTPS for simplicity, gRPC for low-latency communication, and MQTT or message buses for asynchronous, event-driven interactions — based on system requirements.
• Discovery
  Agents need a way to find and connect with each other. A2A supports discovery through mechanisms similar to DNS (e.g., static URLs) or through centralized agent registries — especially useful in enterprise or multi-agent platform settings.
• Security and authentication
  The A2A protocol ensures secure agent communication through authentication (using API keys, OAuth, or identity assertions), authorization for access control, and encryption (like TLS) to protect sensitive data during transmission.

A2A vs MCP

Source: A Visual Guide to Agent2Agent (A2A) Protocol

The Agent-to-Agent (A2A) protocol is designed to facilitate collaboration among multiple AI agents. It enables agents to interact securely through tasks, message exchanges, and artifact sharing. These interactions are stateful, allowing complex workflows to unfold over time. A2A supports discovery via JSON-based AgentCards, letting agents or clients locate and communicate with other agents based on capabilities. This protocol encourages modular, decentralized design where agents from different frameworks (like CrewAI or LlamaIndex) can work together as part of a team.

In contrast, the Model Context Protocol (MCP) is focused on enabling AI agents to access tools, plugins, or APIs. It allows a single model to invoke external capabilities like calculators, data retrievers, or code execution environments. MCP is less about inter-agent collaboration and more about augmenting an individual agent’s power through tool access. Importantly, A2A and MCP aren’t competitors — they’re complementary. A2A agents can be exposed as MCP tools, allowing models using MCP to discover and communicate with other agents via A2A. This layered integration bridges tool invocation with multi-agent orchestration, enabling richer, more intelligent systems.

Experimenting with A2A

In this section, we will explore how to build A2A (Agent-to-Agent) applications. There are three main approaches we can follow:

1. Implementing A2A from scratch using FastAPI,
2. Using Google’s A2A SDK (google-a2a),
3. Leveraging the Python-A2A library, a comprehensive implementation of Google's A2A protocol.

We will begin by building an A2A implementation from scratch using FastAPI to understand the core concepts. Then, we will explore how to create A2A agents using the official Google A2A SDK. Finally, we will use the Python A2A library, which simplifies the development process and provides a robust interface for enabling seamless communication and collaboration between AI agents.

1. Implementing A2A from scratch using FastAPI

Let’s begin with a basic echo agent serves as the “Hello World” of A2A, helping you learn the core concepts by returning whatever input it receives.

Installing uv

We will use uv, a fast and modern Python project manager, to set up and manage our environment. It simplifies tasks like handling dependencies, creating virtual environments, and running scripts.

To install uv, run this in your terminal:

# For Windows
powershell -ExecutionPolicy ByPass -c "irm https://astral.sh/uv/install.ps1 | iex"
set Path=C:\Users\Codem\.local\bin;%Path%

# For linux / Mac
curl -LsSf https://astral.sh/uv/install.sh | sh

Refer to the official website for detailed installation instructions.

Installation | uv

Installing the dependencies

• Initialize a uv project by executing the following command.

uv init basic_a2a_demo
cd basic_a2a_demo

• Create and activate a virtual environment by executing the following command.

uv venv
source .venv/bin/activate # for linux
.venv\Scripts\activate    # for windows

• Install fastapi, uvicorn, requests, uuid and sseclient-py using uv.

uv add fastapi uvicorn requests uuid sseclient-py

Creating an agent card

The agent.json file acts as your agent’s identity card, describing its purpose and how others can interact with it.

Create a file named agent.json and add the following code to it.

{
  "schema_version": "1.0.0",
  "name": "Echo Agent",
  "description": "I repeat what you say, like a friendly cave.",
  "contact_email": "you@example.com",
  "capabilities": [
    "a2a.text-chat"
  ],
  "versions": [
    {
      "version": "1.0.0",
      "endpoint": "http://localhost:8000/a2a",
      "supports_streaming": true,
      "auth": {
        "type": "none"
      }
    }
  ]
}

Building A2A server

Build a basic server that listens for messages and sends them back just as it received like a digital echo.

Create a file named echo_server.py and add the following code to it.

from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse
from sse_starlette.sse import EventSourceResponse
import uuid
import json
import asyncio
from datetime import datetime

app = FastAPI()
# Serve your business card at the standard location
@app.get("/.well-known/agent.json")
async def get_agent_card():
    with open("agent.json") as f:
        return json.load(f)

# Handle regular (non-streaming) requests
@app.post("/a2a/tasks/send")
async def tasks_send(request: Request):
    data = await request.json()
    task_id = data.get("task_id", str(uuid.uuid4()))
    user_message = next((m for m in data.get("messages", []) 
                        if m.get("role") == "user"), None)
    
    if not user_message:
        return JSONResponse(status_code=400, content={"error": "No user message found"})

    parts = user_message.get("parts", [])
    text_parts = [p.get("text") for p in parts if p.get("type") == "text"]
    echo_text = f"Echo: {' '.join(text_parts)}"
    
    return {
        "task_id": task_id,
        "status": "completed",
        "created_time": datetime.utcnow().isoformat(),
        "updated_time": datetime.utcnow().isoformat(),
        "messages": [
            {
                "role": "agent",
                "parts": [{"type": "text", "text": echo_text}]
            }
        ]
    }

if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8000)

Creating A2A client

Create a client to communicate with the server like crafting a remote control for your freshly built device.

import requests
import uuid

class SimpleA2AClient:
    def __init__(self, server_url):
        self.server_url = server_url
        
    def discover_agent(self):
        response = requests.get(f"{self.server_url}/.well-known/agent.json")
        response.raise_for_status()
        return response.json()
        
    def send_message(self, text):
        task_id = str(uuid.uuid4())
        
        # Prepare our request in A2A format
        payload = {
            "task_id": task_id,
            "messages": [
                {
                    "role": "user",
                    "parts": [
                        {
                            "type": "text",
                            "text": text
                        }
                    ]
                }
            ]
        }
        
        # Send the request to the agent
        endpoint = f"{self.server_url}/a2a/tasks/send"
        response = requests.post(endpoint, json=payload)
        response.raise_for_status()
        return response.json()

if __name__ == "__main__":
    # Create our client
    client = SimpleA2AClient("http://localhost:8000")
    
    # Check out the agent's capabilities
    agent_card = client.discover_agent()
    print(f"Found agent: {agent_card['name']}")
    
    # Send a message
    message = input("Type a message to send: ")
    response = client.send_message(message)
    
    # Extract and display the agent's response
    agent_message = response.get("messages", [{}])[0]
    agent_parts = agent_message.get("parts", [{}])
    response_text = agent_parts[0].get("text", "No response") if agent_parts else "No parts"
    
    print(f"\nAgent response: {response_text}")

Executing the app

• Open two terminal windows.
• In the first terminal, start the Echo Server:

python echo_server.py

• In the second terminal, run the client:

python echo_client.py

The output will look like this,

2. Implementing A2A using Google A2A SDK

The A2A Protocol provides a standardized framework that allows agents to work together intelligently and efficiently. The Echo example implemented in the last section highlights the core concepts of A2A, however real-world applications can extend this by connecting agents to language models, databases, and APIs, enabling streaming for real-time updates, adding authentication for secure communication, and coordinating multiple agents to solve complex tasks. Since building these features from scratch is challenging, developers often rely on libraries like Agent Developer Kit (ADK) — Google ADK, Google A2A and Python A2A to simplify the development process.

Installing the dependencies

• Initialize a uv project by executing the following command.

uv init google_a2a_demo
cd google_a2a_demo

• Create and activate a virtual environment by executing the following command.

uv venv
source .venv/bin/activate # for linux
.venv\Scripts\activate    # for windows

• Install a2a-sdk and uvicorn using uv.

uv add a2a-sdk uvicorn

Creating Agent executor

To handle tasks, we need to create an Agent Executor. In a real-world application, this would involve connecting to an LLM or executing other complex logic. For our “Hello World” example, we’ll implement a minimal handler: whenever the agent receives a hello_world task, it simply responds with “Hello, world!”.

The A2A SDK provides an AgentExecutor class, where you define the logic for each skill. For this example, it’s as simple as implementing a function that returns the string "Hello, world" when called.

Create a file named agent_executor.py and add the following code to it.

from a2a.server.agent_execution import AgentExecutor, RequestContext
from a2a.server.events import EventQueue
from a2a.utils import new_agent_text_message

class HelloWorldAgent:
    async def invoke(self) -> str:
        return 'Hello World'

class HelloWorldAgentExecutor(AgentExecutor):
    def __init__(self):
        self.agent = HelloWorldAgent()

    async def execute(
        self,
        context: RequestContext,
        event_queue: EventQueue,
    ) -> None:
        result = await self.agent.invoke()
        await event_queue.enqueue_event(new_agent_text_message(result))

    async def cancel(
        self, context: RequestContext, event_queue: EventQueue
    ) -> None:
        raise Exception('cancel not supported')

Setting up the A2A server

Lets create a simple “Hello World” A2A (Agent-to-Agent) server using the A2A SDK. It defines basic skills, configures the agent’s public and extended profiles, and launches the server to handle requests via HTTP.

Add the following content to main.py file.

import uvicorn

from a2a.server.apps import A2AStarletteApplication
from a2a.server.request_handlers import DefaultRequestHandler
from a2a.server.tasks import InMemoryTaskStore
from a2a.types import AgentCapabilities, AgentCard, AgentSkill
from agent_executor import HelloWorldAgentExecutor

if __name__ == '__main__':
    skill = AgentSkill(
        id='hello_world',
        name='Returns hello world',
        description='just returns hello world',
        tags=['hello world'],
        examples=['hi', 'hello world'],
    )

    extended_skill = AgentSkill(
        id='super_hello_world',
        name='Returns a SUPER Hello World',
        description='A more enthusiastic greeting, only for authenticated users.',
        tags=['hello world', 'super', 'extended'],
        examples=['super hi', 'give me a super hello'],
    )

    # This will be the public-facing agent card
    public_agent_card = AgentCard(
        name='Hello World Agent',
        description='Just a hello world agent',
        url='http://localhost:9999/',
        version='1.0.0',
        default_input_modes=['text'],
        default_output_modes=['text'],
        capabilities=AgentCapabilities(streaming=True),
        skills=[skill],
        supports_authenticated_extended_card=True,
    )

    # This will be the authenticated extended agent card
    specific_extended_agent_card = public_agent_card.model_copy(
        update={
            'name': 'Hello World Agent - Extended Edition',  # Different name for clarity
            'description': 'The full-featured hello world agent for authenticated users.',
            'version': '1.0.1',  # Could even be a different version
            'skills': [
                skill,
                extended_skill,
            ],
        }
    )

    request_handler = DefaultRequestHandler(
        agent_executor=HelloWorldAgentExecutor(),
        task_store=InMemoryTaskStore(),
    )

    server = A2AStarletteApplication(
        agent_card=public_agent_card,
        http_handler=request_handler,
        extended_agent_card=specific_extended_agent_card,
    )

    uvicorn.run(server.build(), host='0.0.0.0', port=9999)

In this code,

• we define a skill (AgentSkill) that returns a simple "Hello, world!" message, along with an extended version for authenticated users.
• An agent card (AgentCard) describes the agent’s capabilities, including its skills and supported input/output modes. A separate extended agent card provides enhanced functionality for authenticated clients.
• The request handler connects the logic (HelloWorldAgentExecutor) to an in-memory task store.
• Finally, we use A2AStarletteApplication (built on Starlette) to expose the agent as an HTTP service and run it using Uvicorn on port 9999.

Setting up the client

Lets sets up an HTTP client, fetches the agent’s card (including an optional extended version for authenticated access), initializes the A2A client, and sends a message query ("How much is 10 USD in INR?") both as a regular and a streaming message.

Create a file named test_client.py and add the following code to it.

import logging
from typing import Any
from uuid import uuid4
import httpx

from a2a.client import A2ACardResolver, A2AClient
from a2a.types import (
    AgentCard,
    MessageSendParams,
    SendMessageRequest,
    SendStreamingMessageRequest,
)


async def main() -> None:
    PUBLIC_AGENT_CARD_PATH = '/.well-known/agent.json'
    EXTENDED_AGENT_CARD_PATH = '/agent/authenticatedExtendedCard'

    logging.basicConfig(level=logging.INFO)
    logger = logging.getLogger(__name__)  

    base_url = 'http://localhost:9999'

    async with httpx.AsyncClient() as httpx_client:
        # Initialize A2ACardResolver
        resolver = A2ACardResolver(
            httpx_client=httpx_client,
            base_url=base_url,
        )
        # Fetch Public Agent Card and Initialize Client
        final_agent_card_to_use: AgentCard | None = None

        try:
            logger.info(f'Attempting to fetch public agent card from: {base_url}{PUBLIC_AGENT_CARD_PATH}')
            _public_card = await resolver.get_agent_card()
            logger.info('Successfully fetched public agent card:')
            logger.info(_public_card.model_dump_json(indent=2, exclude_none=True))
            final_agent_card_to_use = _public_card
            logger.info('\nUsing PUBLIC agent card for client initialization (default).')

            if _public_card.supports_authenticated_extended_card:
                try:
                    logger.info(f'\nPublic card supports authenticated extended card. Attempting to fetch from: {base_url}{EXTENDED_AGENT_CARD_PATH}')
                    auth_headers_dict = {
                        'Authorization': 'Bearer dummy-token-for-extended-card'
                    }
                    _extended_card = await resolver.get_agent_card(
                        relative_card_path=EXTENDED_AGENT_CARD_PATH,
                        http_kwargs={'headers': auth_headers_dict},
                    )
                    logger.info('Successfully fetched authenticated extended agent card:')
                    logger.info(
                        _extended_card.model_dump_json(
                            indent=2, exclude_none=True
                        )
                    )
                    final_agent_card_to_use = _extended_card
                    logger.info('\nUsing AUTHENTICATED EXTENDED agent card for client initialization.')
                except Exception as e_extended:
                    logger.warning(
                        f'Failed to fetch extended agent card: {e_extended}. Will proceed with public card.',
                        exc_info=True,
                    )
            elif (
                _public_card
            ):  # supportsAuthenticatedExtendedCard is False or None
                logger.info('\nPublic card does not indicate support for an extended card. Using public card.')

        except Exception as e:
            logger.error(f'Critical error fetching public agent card: {e}', exc_info=True)
            raise RuntimeError(
                'Failed to fetch the public agent card. Cannot continue.'
            ) from e

        client = A2AClient(
            httpx_client=httpx_client, agent_card=final_agent_card_to_use
        )
        logger.info('A2AClient initialized.')

        send_message_payload: dict[str, Any] = {
            'message': {
                'role': 'user',
                'parts': [
                    {'kind': 'text', 'text': 'How much is 10 USD in INR?'}
                ],
                'messageId': uuid4().hex,
            },
        }
        request = SendMessageRequest(
            id=str(uuid4()), params=MessageSendParams(**send_message_payload)
        )
        response = await client.send_message(request)
        print(response.model_dump(mode='json', exclude_none=True))

        streaming_request = SendStreamingMessageRequest(
            id=str(uuid4()), params=MessageSendParams(**send_message_payload)
        )
        stream_response = client.send_message_streaming(streaming_request)

        async for chunk in stream_response:
            print(chunk.model_dump(mode='json', exclude_none=True))

if __name__ == '__main__':
    import asyncio

    asyncio.run(main())

In this code,

• An asynchronous HTTP client (httpx.AsyncClient) is used to fetch a public agent card from a locally running A2A server. It then attempts to retrieve an extended, authenticated version of the card using a dummy bearer token.
• With the obtained card (either public or extended), the script initializes an A2AClient, which is then used to send a currency conversion query ("how much is 10 USD in INR?").
• The response is printed using both the standard send_message method and the streaming send_message_streaming method, showcasing how to handle real-time agent replies.

A2A Client-Server Interaction Flow

Image source: A2A Samples: Hello World Agent | A2A Protocol

Executing the app

• Open two terminal windows.
• In the first terminal, start the A2A Server:

uv run main.py

• In the second terminal, run the client:

python test_client.py

The output will look like this,

3. Implementing A2A using Python-A2A library

The Python-A2A library is a powerful and developer-friendly extension built on top of Google’s Agent-to-Agent (A2A) protocol. While Google’s native A2A SDK provides foundational tools to build interoperable agents and define standardized communication formats. It is ideal for developers looking to quickly prototype, deploy, and scale intelligent agents that can autonomously communicate and collaborate using the A2A standard. It bridges the gap between low-level protocol details and high-level use cases.

Installing the dependencies

• Initialize a uv project by executing the following command.

uv init python_a2a_demo
cd python_a2a_demo

• Create and activate a virtual environment by executing the following command.

uv venv
source .venv/bin/activate # for linux
.venv\Scripts\activate    # for windows

• Install python-a2a and python-dotenv using uv.

uv add python-a2a[all] python-dotenv

Example 1: Echo Agent

Lets begin with building a simple A2A compatible agent using the python-a2a library. The agent, named "Echo Agent", is designed to echo user's messages.

Creating echo agent server
Create a file named echo_agent.py and add the following code to create a basic A2A agent server.

from python_a2a import A2AServer, Message, TextContent, MessageRole, run_server

class EchoAgent(A2AServer):
    def handle_message(self, message):
        if message.content.type == "text":
            return Message(
                content=TextContent(text=f"Echo: {message.content.text}"),
                role=MessageRole.AGENT,
                parent_message_id=message.message_id,
                conversation_id=message.conversation_id
            )

if __name__ == "__main__":
    agent = EchoAgent()
    run_server(agent, host="0.0.0.0", port=5000)

In this code,

• Defines an EchoAgent class by extending A2AServer to handle incoming A2A protocol messages.
• When a text message is received, it responds with an echo by prepending "Echo:" to the input.
• The response maintains conversation context using message and conversation IDs.
• Launches the agent using run_server on host 0.0.0.0 and port 5000, making it accessible for incoming agent-to-agent communication.

Creating echo client
Create a file named echo_client.py and add the following code to create a client for the echo agent.

from python_a2a import A2AClient, Message, TextContent, MessageRole

client = A2AClient("http://localhost:5000/a2a")
message = Message(
    content=TextContent(text="Hello, Good morning!"),
    role=MessageRole.USER
)
response = client.send_message(message)
print(f"Agent says: {response.content.text}")

In this code,

• Creates an A2AClient that connects to an agent running at http://localhost:5000/a2a.
• Constructs a user message with the content "Hello, Good morning!" using the A2A message format.
• Sends the message to the agent using client.send_message(...).
• Receives and prints the agent’s response, showing the echoed reply from the server.

Executing the code
Open separate terminals in the project folder, activate the virtual environment in each, and run the agent server and client separately to execute the code.

# terminal 1
python echo_agent.py
# terminal 2
python echo_client.py

The output will look like the following:

Example 2: Basic A2A Agent

Lets begin with building a simple A2A compatible agent using the python-a2a library. The agent, named "Greeting Agent", is designed to detect greetings in a user's message and respond accordingly.

Creating greeting agent server
Create a file named greeting_agent.py and add the following code to create a basic A2A agent server.

from python_a2a import A2AServer, skill, agent, run_server
from python_a2a import TaskStatus, TaskState

@agent(
    name="Greeting Agent",
    description="A simple agent that responds to greetings",
    version="1.0.0"
)
class GreetingAgent(A2AServer):

    @skill(
        name="Greet",
        description="Respond to a greeting",
        tags=["greeting", "hello"]
    )
    def greet(self, name=None):
        if name:
            return f"Hello, {name}! How can I help you today?"
        else:
            return "Hello there! How can I help you today?"

    def handle_task(self, task):
        message_data = task.message or {}
        content = message_data.get("content", {})
        text = content.get("text", "") if isinstance(content, dict) else ""

        greeting_words = ["hello", "hi", "hey", "greetings"]
        is_greeting = any(word in text.lower() for word in greeting_words)

        if is_greeting:
            name = None
            if "my name is" in text.lower():
                name = text.lower().split("my name is")[1].strip()

            greeting = self.greet(name)
            task.artifacts = [{
                "parts": [{"type": "text", "text": greeting}]
            }]
            task.status = TaskStatus(state=TaskState.COMPLETED)
        else:
            task.artifacts = [{
                "parts": [{"type": "text", "text": "I'm a greeting agent. Try saying hello!"}]
            }]
            task.status = TaskStatus(state=TaskState.COMPLETED)

        return task

# Run the server
if __name__ == "__main__":
    agent = GreetingAgent()
    run_server(agent, port=5000)

In this code,

• The agent is defined using the @agent decorator, and a skill named Greet is added with the @skill decorator to handle greeting responses.
• The handle_task method identifies whether the incoming message is a greeting and responds accordingly, using the skill defined.
• Finally, the server runs the agent on port 5000 using run_server.

Creating greeting agent client
Create a file named greeting_client.py and add the following code to create a client for the greeting agent.

from python_a2a import A2AClient

# Create a client
client = A2AClient("http://localhost:5000")

# Print agent information
print(f"Connected to: {client.agent_card.name}")
print(f"Description: {client.agent_card.description}")
print(f"Skills: {[skill.name for skill in client.agent_card.skills]}")

# Send a greeting
response = client.ask("Hello there! My name is Vishnu.")
print(f"Response: {response}")

# Send another message
response = client.ask("What can you do?")
print(f"Response: {response}")

Executing the code
Open separate terminals in the project folder, activate the virtual environment in each, and run the agent server and client separately to execute the code.

# terminal 1
python greeting_agent.py
# terminal 2
python greeting_client.py

The output will look like the following:

Example 3: LLM-Based Agent

Lets build an OpenAI-powered Agent-to-Agent (A2A) server using the python-a2a library. The agent uses OpenAI's GPT model to respond to queries.

Create .env File
In your project root directory, create a file named .env with the following content:

OPENAI_API_KEY=your_openai_api_key_here

Creating LLM Server
Create a file named llm_agent.py and add the following code to it.

from python_a2a import OpenAIA2AServer, run_server

import os
from dotenv import load_dotenv
load_dotenv()
api_key = os.environ.get("OPENAI_API_KEY")
if not api_key:
    raise ValueError("Please set the OPENAI_API_KEY environment variable.")

# Create an OpenAI-based A2A agent
agent = OpenAIA2AServer(
    api_key=api_key,
    model="gpt-4",
    system_prompt="You are a helpful assistant that specializes in explaining complex concepts simply."
)

# Run the server
if __name__ == "__main__":
    print("Starting OpenAI-based A2A agent...")
    run_server(agent, host="0.0.0.0", port=5000)

Creating LLM Client
Create a file named llm_client.py and add the following code to it.

from python_a2a import A2AClient

# Connect to the OpenAI-based A2A agent
client = A2AClient("http://localhost:5000")

# Print agent metadata
print(f"Connected to: {client.agent_card.name}")
print(f"Description: {client.agent_card.description}")
print("Skills:")
for skill in client.agent_card.skills:
    print(f" - {skill.name}: {skill.description}")

# Send some example questions to test
messages = [
    "In short, why speed of light is constant?",
    "Explain bell Inequality experiment in simple terms.",
]

# Send messages and print responses
for msg in messages:
    response = client.ask(msg)
    print(f"\nUser: {msg}")
    print(f"Agent: {response}")

Executing the code
Open separate terminals in the project folder, activate the virtual environment in each, and run the agent server and client separately to execute the code.

# terminal 1
python llm_agent.py
# terminal 2
python llm_client.py

The output will look like the following:

Example 4: Converting LangChain to A2A Servers

You can turn any LangChain agent or chain into an A2A-compatible server.

LangChain requires a few additional libraries for proper integration. Run the following code to install the necessary dependencies into the project.

uv add langchain-community langchain-openai numexpr

Create a file named langchain_to_a2a_server.py and add the following code to it.

from langchain.chains import LLMMathChain
from langchain_openai import OpenAI
from python_a2a.langchain import to_a2a_server
from python_a2a import run_server

import os
from dotenv import load_dotenv
load_dotenv()
api_key = os.environ.get("OPENAI_API_KEY")
if not api_key:
    raise ValueError("Please set the OPENAI_API_KEY environment variable.")

# Create a LangChain chain
llm = OpenAI(temperature=0)
math_chain = LLMMathChain(llm=llm)

# Convert to A2A server
a2a_server = to_a2a_server(math_chain)

# Run the server
if __name__ == "__main__":
    run_server(a2a_server, port=5000)

You can reuse the previously created llm_client for this agent server using the following code. Execute both server and client to see the result.

The output will look like the following:

Example 5: Converting MCP Tools to LangChain Tools

You can convert any MCP tools into LangChain-compatible tools, enabling them to be used directly with LangChain agents.

from python_a2a.mcp import FastMCP
from python_a2a.langchain import to_langchain_tool
from python_a2a import run_server
from langchain.agents import initialize_agent, AgentType
from langchain.llms import OpenAI

# Create an MCP server with tools
calculator = FastMCP(name="Calculator MCP")

@calculator.tool()
def add(a: float, b: float) -> float:
    """Add two numbers together."""
    return a + b

@calculator.tool()
def subtract(a: float, b: float) -> float:
    """Subtract b from a."""
    return a - b

# Run the MCP server in a background thread
import threading
server_thread = threading.Thread(
    target=run_server,
    args=(calculator,),
    kwargs={"port": 8000},
    daemon=True
)
server_thread.start()

# Convert MCP tools to LangChain tools
add_tool = to_langchain_tool("http://localhost:8000", "add")
subtract_tool = to_langchain_tool("http://localhost:8000", "subtract")

# Use in a LangChain agent
llm = OpenAI(temperature=0)
tools = [add_tool, subtract_tool]

agent = initialize_agent(
    tools, 
    llm, 
    agent=AgentType.ZERO_SHOT_REACT_DESCRIPTION,
    verbose=True
)

# Run the agent
result = agent.run("Add 15 and 27, then subtract 5 from the result")
print(result)

Building a travel planner app using A2A

In this section, we will build a multi-agent Travel Planner powered by the A2A (Agent-to-Agent) protocol. This system leverages LangChain, the Ollama LLM, and multiple A2A-compatible agents that work collaboratively to retrieve weather information, perform web searches, and generate a comprehensive travel recommendation.

Application flow

1. The user requests a travel plan (e.g., “Plan a trip to Kerala”).
2. The Travel Planner asks the Weather Agent for the forecast.
3. Based on the weather (clear or rainy), it decides the activity type.
4. It then queries the Tavily Search Agent for suitable activities.
5. A local LLM (e.g., via Ollama) compiles this into a final itinerary — ensuring user privacy without third-party data sharing.

Prerequisites

This hands-on requires the following tools to be installed on your machine:

1. Ollama: Ollama is a platform for running large language models locally on your computer.

Download Ollama on Windows

Run the following command in your terminal to pull the model using Ollama:

ollama pull llama3.2

2. Python: Python is the core language used in this hands-on for scripting and backend logic.

Download Python

3. uv (Micro virtualenv manager): uv is a fast and modern Python project manager, to set up and manage our environment.

To install uv, run this in your terminal:

# For Windows
powershell -ExecutionPolicy ByPass -c "irm https://astral.sh/uv/install.ps1 | iex"
set Path=C:\Users\Codem\.local\bin;%Path%

# For linux / Mac
curl -LsSf https://astral.sh/uv/install.sh | sh

Refer to the official website for detailed installation instructions.

Installation | uv

Installing the dependencies

• Initialize a uv project by executing the following command.

uv init travel_planner
cd travel_planner

• Create and activate a virtual environment by executing the following command.

uv venv
source .venv/bin/activate # for linux
.venv\Scripts\activate    # for windows

• Install python-a2a, langchain-ollama, python-dotenv, tavily-python and streamlit using uv.

uv add python-a2a langchain-ollama python-dotenv tavily-python streamlit

Setting up the environment

This hands-on project uses Tavily and OpenWeather API keys.

• Visit the Tavily official website and sign in to obtain your API key.
• Go to the OpenWeather website and create a new account. Once you complete the sign-up, you will receive your API key in your mail.
• In the root directory of your project, create a .env file and add the following content:

OPENWEATHER_API_KEY=fedc837bbb7477...
TAVILY_API_KEY=tvly-dev-h5xKcqcytBeJQ...

Creating Weather Agent

Create a file named WeatherAgent.py and add the following code to it.

from python_a2a import A2AServer, skill, agent, run_server, TaskStatus, TaskState

import os
import requests
import logging

from dotenv import load_dotenv
load_dotenv()
api_key = os.environ.get("OPENWEATHER_API_KEY")


@agent(
    name="Weather Agent",
    description="Provides weather information",
    version="1.0.0",
    url="https://zzz.example.com"
)
class WeatherAgent(A2AServer):
    
    @skill(
        name="Get Weather",
        description="Get current weather for a location",
        tags=["weather", "forecast"],
        examples="I am a weather agent for getting weather forecast from Open weather"
    )
    def get_weather(self, location):
        if not api_key:
            return "Weather service not available (missing API key)."
        
        try:
            url = (
                f"https://api.openweathermap.org/data/2.5/weather?"
                f"q={location}&units=imperial&appid={api_key}"
            )
            logging.debug(f"Request URL: {url}")  # Log the full request URL

            response = requests.get(url, timeout=5)
            response.raise_for_status()
            logging.debug(f"Response Status Code: {response.status_code}")  # Log status code
            logging.debug(f"Response Text: {response.text}")  # Log raw response text
            
            data = response.json()
            
            temp = data["main"]["temp"]
            description = data["weather"][0]["description"]
            city_name = data["name"]

            logging.debug(f"Parsed Data: Temp = {temp}, Description = {description}, City = {city_name}")
            
            return f"The weather in {city_name} is {description} with a temperature of {temp}°F."
        
        except requests.RequestException as e:
            return f"Error fetching weather: {e}"
        except (KeyError, TypeError):
            return "Could not parse weather data."
    
    def handle_task(self, task):
        # Extract location from message
        message_data = task.message or {}
        content = message_data.get("content", {})
        text = content.get("text", "") if isinstance(content, dict) else ""
        
        if "weather" in text.lower() and "in" in text.lower():
            location = text.split("in", 1)[1].strip().rstrip("?.")
            
            # Get weather and create response
            weather_text = self.get_weather(location)
            task.artifacts = [{
                "parts": [{"type": "text", "text": weather_text}]
            }]
            task.status = TaskStatus(state=TaskState.COMPLETED)
        else:
            task.status = TaskStatus(
                state=TaskState.INPUT_REQUIRED,
                message={"role": "agent", "content": {"type": "text", 
                         "text": "Please ask about weather in a specific location."}}
            )
        return task

# Run the server
if __name__ == "__main__":
    agent = WeatherAgent(google_a2a_compatible=True)
    run_server(agent, port=8001, debug=True)

In this code,

• Loads the OpenWeather API key from .env and defines a weather agent using A2AServer.
• Registers a skill to fetch and return weather data for a given location using OpenWeatherMap API.
• Handles user tasks by extracting location from the message and invoking the weather skill.
• Responds with the weather info or prompts the user if the location is unclear.
• Runs the agent server on port 8001 with debug mode enabled.

Creating Tavily Search Agent

Create a file named TavilySearchAgent.py and add the following code to it.

from python_a2a import A2AServer, skill, agent, run_server, TaskStatus, TaskState
from tavily import TavilyClient
import os
import logging

from dotenv import load_dotenv
load_dotenv()
api_key = os.environ.get("TAVILY_API_KEY")

@agent(
    name="Tavily Search Agent",
    description="Performs internet search using Tavily API",
    version="1.0.0",
    url="https://yourdomain.com"
)
class TavilySearchAgent(A2AServer):

    def __init__(self, **kwargs):
        super().__init__(**kwargs)
        self.client = TavilyClient(api_key)

    @skill(
        name="Search Internet",
        description="Perform a web search using Tavily API",
        tags=["search", "internet", "tavily"],
        examples="Search 'must visit places in utah in may'"
    )
    def search(self, query: str):
        """Perform search using Tavily Search API"""
        try:
            response = self.client.search(query=query)

            results = response.get("results", [])
            if not results:
                return "No search results found."

            summary = "\n".join(
                [f"- {r.get('title')}: {r.get('url')}" for r in results]
            )
            return f"Top results for '{query}':\n{summary}"

        except Exception as e:
            logging.error(f"Error during Tavily search: {e}")
            return f"Search failed: {e}"

    def handle_task(self, task):
        message_data = task.message or {}
        content = message_data.get("content", {})
        text = content.get("text", "") if isinstance(content, dict) else ""

        if text.strip():
            query = text.strip()
            result = self.search(query)
            task.artifacts = [{
                "parts": [{"type": "text", "text": result}]
            }]
            task.status = TaskStatus(state=TaskState.COMPLETED)
        else:
            task.status = TaskStatus(
                state=TaskState.INPUT_REQUIRED,
                message={"role": "agent", "content": {"type": "text", 
                         "text": "Please provide a search query."}}
            )
        return task


if __name__ == "__main__":
    agent = TavilySearchAgent(google_a2a_compatible=True)
    run_server(agent, port=8002, debug=True)

Creating Local LLM Agent

Create a file named LocalLLMAgent.py and add the following code to it.

from python_a2a import run_server
from python_a2a.langchain import to_a2a_server
from langchain_ollama.llms import OllamaLLM

# Create a LangChain LLM
#llm = ChatOpenAI(model="gpt-3.5-turbo", temperature=0)
llm = OllamaLLM(model="llama3.2:latest")

# Convert LLM to A2A server
llm_server = to_a2a_server(llm)

if __name__ == "__main__":
    print("Starting LLM A2A server on port 5001...")
    run_server(llm_server, port=5001)

Creating Travel Planner Agent

Create a file named TravelPlannerApp.py and add the following code to it.

import streamlit as st
from python_a2a import AgentNetwork, A2AClient
import asyncio

# Function to run async logic inside Streamlit
def run_async(coro):
    return asyncio.run(coro)

async def plan_trip(destination, travel_dates):
    # Create an agent network
    network = AgentNetwork(name="Travel Assistant Network")
    network.add("weather", "http://localhost:8001")
    network.add("search", "http://localhost:8002")

    # Get agents
    weather_agent = network.get_agent("weather")
    search_agent = network.get_agent("search")
    llm_client = A2AClient("http://localhost:5001")

    # Get weather forecast
    forecast = weather_agent.ask(f"What's the weather in {destination}?")

    # Search based on weather
    if "sunny" in forecast.lower() or "clear" in forecast.lower():
        activities = search_agent.ask(f"Recommend outdoor activities in {destination}")
    else:
        activities = search_agent.ask(f"Recommend indoor activities in {destination}")

    # Summarize using LLM
    prompt = (
        f"You are a travel assistant. Based on the weather forecast result '{forecast}' "
        f"and the recommendations [{activities}], suggest me a few must-see attractions "
        f"on date {travel_dates}."
    )

    llm_result = llm_client.ask(prompt)

    return forecast, activities, llm_result

# Streamlit UI
st.set_page_config(page_title="🧳 Travel Planner Assistant")

st.title("🧭 Travel Planner Assistant")
st.write("Get personalized trip suggestions based on real-time weather and recommendations.")

destination = st.text_input("Enter destination", value="Kerala, India")
travel_dates = st.text_input("Enter travel dates", value="August 1-5")

if st.button("Plan My Trip"):
    with st.spinner("Planning your trip..."):
        try:
            forecast, activities, llm_result = run_async(plan_trip(destination, travel_dates))

            st.subheader("📍 Weather Forecast")
            st.success(forecast)

            st.subheader("🎯 Recommended Activities")
            st.info(activities)

            st.subheader("🗺️ Suggested Travel Plan")
            st.markdown(llm_result)
        except Exception as e:
            st.error(f"Something went wrong: {e}")

In this code,

• Initializes a Streamlit web UI for a travel planner that collects user inputs like destination and travel dates.
• Defines an asynchronous function to coordinate multiple agents via the AgentNetwork class.
• Connects to weather, search, and LLM agents running locally on different ports.
• Based on the weather forecast for the destination, selects indoor or outdoor activity recommendations.
• Summarizes the final travel plan using a language model (LLM) agent and displays the results on the UI.
• Uses asyncio.run() to integrate asynchronous agent responses within Streamlit’s synchronous workflow.

Creating the executor

You can either run each agent separately or create an executor script that uses Python’s subprocess module to launch all agents sequentially.

In this hands-on exercise, we will follow the subprocess approach to simplify execution.

Open your main.py file and replace its contents with the following code:

import subprocess
import time
import sys

def main():
    print("Hello from travel-planner!")

    # Scripts to launch before the Streamlit app
    scripts = ["WeatherAgent.py", "TavilySearchAgent.py", "LocalLLMAgent.py"]
    streamlit_app = "TravelPlannerApp.py"

    processes = []

    # Launch agent scripts
    for script in scripts:
        print(f"Launching {script}...")
        p = subprocess.Popen([sys.executable, script])
        processes.append(p)
        print(f"{script} started. Waiting 2 seconds before next...\n")
        time.sleep(2)

    # Launch Streamlit app
    print(f"Launching Streamlit app: {streamlit_app}...")
    p = subprocess.Popen(["streamlit", "run", streamlit_app])
    processes.append(p)

    # Keep the main process alive
    try:
        print("All agents (and UI) are running. Press Ctrl+C to stop.")
        while True:
            time.sleep(1)
    except KeyboardInterrupt:
        print("\nShutting down all agents...")
        for p in processes:
            p.terminate()
        print("All agents stopped.")

if __name__ == "__main__":
    main()

Executing the app

To launch the entire application using subprocess, simply run the following command:

uv run main.py

If you would prefer to run each agent separately, open four terminal windows and execute the following commands in each terminal:

python WeatherAgent.py
python TavilySearchAgent.py
python LocalLLMAgent.py
streamlit run TravelPlannerApp.py

🎉 Awesome Work! You’ve successfully built a Travel Planner powered by the A2A protocol.

Thanks for reading this article !!

Thanks Gowri M Bhatt for reviewing the content.

If you enjoyed this article, please click on the clap button 👏 and share to help others find it!

The full source code for this tutorial can be found here,

GitHub - codemaker2015/a2a-experiments: This repository contains experiments and demos for…

GitHub - codemaker2015/travel-planner: This project is a multi-agent Travel Planner powered by the…

References

A Visual Guide to Agent2Agent (A2A) Protocol

GitHub - a2aproject/a2a-samples: Samples using the Agent2Agent (A2A) Protocol

A2A Protocol Documentation

A2A Samples: Hello World Agent