A2A¶

Agentle provides built-in support for the Agent-to-Agent (A2A) Protocol, enabling standardized communication between agents. This page explains how to use the A2A features in Agentle.

What is A2A?¶

A2A (Agent-to-Agent) is an open protocol designed to enable standardized communication between autonomous agents built on different frameworks and by various vendors. It provides a common interface for agent communication, task management, and state tracking.

Basic A2A Usage¶

Here’s a simple example of using the A2A interface:

import os
import time

from agentle.agents.a2a.a2a_interface import A2AInterface
from agentle.agents.a2a.message_parts.text_part import TextPart
from agentle.agents.a2a.messages.message import Message
from agentle.agents.a2a.tasks.task_query_params import TaskQueryParams
from agentle.agents.a2a.tasks.task_send_params import TaskSendParams
from agentle.agents.a2a.tasks.task_state import TaskState
from agentle.agents.agent import Agent
from agentle.generations.providers.google.google_genai_generation_provider import GoogleGenaiGenerationProvider

# Set up agent and A2A interface
provider = GoogleGenaiGenerationProvider(api_key=os.environ.get("GOOGLE_API_KEY"))
agent = Agent(
    name="Example Agent",
    generation_provider=provider,
    model="gemini-2.0-flash"
)
a2a = A2AInterface(agent=agent)

# Send task to agent
message = Message(
    role="user",
    parts=[TextPart(text="What are three facts about the Moon?")]
)
task = a2a.tasks.send(TaskSendParams(message=message))
print(f"Task sent with ID: {task.id}")

# Wait for task completion and get result
while True:
    result = a2a.tasks.get(TaskQueryParams(id=task.id))
    status = result.result.status

    if status == TaskState.COMPLETED:
        print("\nResponse:", result.result.history[1].parts[0].text)
        break
    elif status == TaskState.FAILED:
        print(f"Task failed: {result.result.error}")
        break
    print(f"Status: {status}")
    time.sleep(1)

A2A Components¶

The A2A implementation in Agentle consists of several key components:

A2AInterface¶

The main entry point for A2A functionality:

from agentle.agents.a2a.a2a_interface import A2AInterface
from agentle.agents.agent import Agent

# Create an agent
agent = Agent(
    name="Travel Agent",
    generation_provider=provider,
    model="gemini-2.0-flash",
    instructions="You are a travel agent that helps plan trips."
)

# Create an A2A interface for the agent
a2a = A2AInterface(agent=agent)

Task Management¶

The task management system provides methods for sending, querying, and canceling tasks:

from agentle.agents.a2a.tasks.task_send_params import TaskSendParams
from agentle.agents.a2a.tasks.task_query_params import TaskQueryParams
from agentle.agents.a2a.tasks.task_cancel_params import TaskCancelParams
from agentle.agents.a2a.message_parts.text_part import TextPart
from agentle.agents.a2a.messages.message import Message

# Create a message
message = Message(
    role="user",
    parts=[TextPart(text="Plan a 3-day trip to Tokyo")]
)

# Send a task
task = a2a.tasks.send(TaskSendParams(message=message))
task_id = task.id

# Query a task
task_result = a2a.tasks.get(TaskQueryParams(id=task_id))

# Cancel a task
cancel_result = a2a.tasks.cancel(TaskCancelParams(id=task_id))

Messages and Parts¶

A2A uses a structured message format with different part types:

from agentle.agents.a2a.messages.message import Message
from agentle.agents.a2a.message_parts.text_part import TextPart
from agentle.agents.a2a.message_parts.file_part import FilePart

# Create a text-only message
text_message = Message(
    role="user",
    parts=[TextPart(text="Analyze this data")]
)

# Create a multimodal message with text and image
with open("chart.png", "rb") as f:
    image_data = f.read()

multimodal_message = Message(
    role="user",
    parts=[
        TextPart(text="Analyze this chart:"),
        FilePart(data=image_data, mime_type="image/png")
    ]
)

Task States¶

The A2A protocol defines several task states for tracking progress:

State	Description
`PENDING`	The task has been received but not yet started
`RUNNING`	The task is currently being processed
`COMPLETED`	The task has completed successfully
`FAILED`	The task encountered an error
`CANCELED`	The task was canceled by the client

Task state transitions are handled automatically by the A2A interface.

Exposing A2A via API¶

For production use, you can expose your A2A interface as a RESTful API:

from agentle.agents.a2a.a2a_interface import A2AInterface
from agentle.agents.agent import Agent
from agentle.agents.asgi.blacksheep.agent_to_blacksheep_application_adapter import AgentToBlackSheepApplicationAdapter
from agentle.generations.providers.google.google_genai_generation_provider import GoogleGenaiGenerationProvider

# Create your agent
travel_agent = Agent(
    name="Travel Agent",
    description="An AI assistant specialized in planning travel itineraries.",
    generation_provider=GoogleGenaiGenerationProvider(),
    model="gemini-2.0-flash",
    instructions="""You are a travel agent specialized in creating detailed
    itineraries and providing travel recommendations.""",
)

# Create an A2A interface for the agent
a2a_interface = A2AInterface(agent=travel_agent)

# Convert the A2A interface to a BlackSheep ASGI application
app = AgentToBlackSheepApplicationAdapter().adapt(a2a_interface)

# Run the API server
if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="127.0.0.1", port=8000)

This creates an API with the following endpoints:

POST /api/v1/tasks/send - Send a task to the agent asynchronously
POST /api/v1/tasks/get - Get task results
POST /api/v1/tasks/cancel - Cancel a running task
WebSocket /api/v1/notifications - Subscribe to push notifications about task status changes

Client-Side A2A Communication¶

To interact with an A2A-compatible service from client code:

import requests
import json
import time

# Define the A2A service endpoint
base_url = "http://localhost:8000/api/v1"

# Create a message
message = {
    "role": "user",
    "parts": [
        {
            "type": "text",
            "text": "Plan a 3-day trip to Tokyo"
        }
    ]
}

# Send a task
send_response = requests.post(
    f"{base_url}/tasks/send",
    json={"message": message}
)
task_id = send_response.json()["id"]
print(f"Task sent with ID: {task_id}")

# Poll for results
while True:
    get_response = requests.post(
        f"{base_url}/tasks/get",
        json={"id": task_id}
    )
    result = get_response.json()
    status = result["result"]["status"]

    if status == "COMPLETED":
        print("\nResponse:", result["result"]["history"][1]["parts"][0]["text"])
        break
    elif status == "FAILED":
        print(f"Task failed: {result['result']['error']}")
        break
    print(f"Status: {status}")
    time.sleep(1)

Benefits of A2A¶

The A2A protocol offers several advantages:

Interoperability: Agents built with different frameworks can communicate seamlessly
Enterprise Integration: Easily integrate agents into existing enterprise applications
Asynchronous Communication: Non-blocking task management for long-running operations
State Management: Track task progress and history across agent interactions
Multimodal Support: Exchange rich content including text, images, and structured data
Open Standard: Community-driven protocol designed for widespread adoption

Implementation Details¶

Agentle’s A2A implementation handles the complexity of:

Task Lifecycle Management: Automatically manages task creation, execution, and state transitions
Thread-Safe Execution: Uses isolated threads with dedicated event loops to prevent concurrency issues
Error Handling: Provides robust error recovery mechanisms during task execution
Standardized Messaging: Offers a clean interface for creating, sending, and processing A2A messages
Session Management: Maintains conversation history and context across multiple interactions
Asynchronous Processing: Transparently converts asynchronous A2A operations into synchronous methods

Advanced A2A Configuration¶

You can customize the A2A interface with additional parameters:

from agentle.agents.a2a.a2a_interface import A2AInterface, A2AInterfaceOptions

# Configure the A2A interface with options
a2a = A2AInterface(
    agent=agent,
    options=A2AInterfaceOptions( # wip
        max_concurrent_tasks=10,  # Maximum number of concurrent tasks
        task_timeout=300,         # Task timeout in seconds
        task_poll_interval=1,     # Polling interval for task status
        keep_completed_tasks=100  # Number of completed tasks to keep in history
    )
)

Best Practices¶

Error Handling: Implement robust error handling for task failures
Timeouts: Set appropriate timeouts for long-running tasks
Scaling: Consider using a message broker for high-volume A2A implementations
Authentication: Add authentication for production A2A endpoints
Monitoring: Monitor task throughput, completion rates, and error rates