A2A Client

Async Python client for communicating with A2A-compliant servers

from praisonaiagents import Agent

agent = Agent(name="a2a-agent", instructions="Communicate with remote A2A agents.")
agent.start("Connect to the remote A2A server and fetch available tasks.")

A2AClient enables PraisonAI applications to discover and interact with remote A2A agents through an async HTTP interface.

The user asks the agent to reach a remote A2A server; the client discovers capabilities and exchanges messages over HTTP.

Quick Start

1. Simple Usage

   Connect to an A2A server and send a message:

   import asyncio
   from praisonaiagents.ui.a2a import A2AClient
   
   async def main():
       async with A2AClient("http://localhost:8000") as client:
           # Discover agent capabilities
           card = await client.get_agent_card()
           print(f"Agent: {card.name}")
           
           # Send a message
           result = await client.send_message("What are AI trends?")
           print(result)
   
   asyncio.run(main())
   
2. With Authentication

   Use bearer token authentication:

   from praisonaiagents.ui.a2a import A2AClient
   
   async with A2AClient(
       base_url="http://localhost:8000",
       auth_token="sk-secret",
       timeout=30.0
   ) as client:
       result = await client.send_message("Hello!")
       print(result)
   

---

How It Works

Step	Description
Discovery	Fetch agent capabilities via Agent Card
Authentication	Optional Bearer token validation
Messaging	Send messages using JSON-RPC protocol
Task Management	Track, list, and cancel tasks

---

Configuration Options

Parameter	Type	Default	Description
base_url	str	—	Base URL of the A2A server
auth_token	str	None	Bearer token for authentication
timeout	float	30.0	HTTP request timeout in seconds

---

Discovery Methods

Get Agent Card

Fetch basic agent information from /.well-known/agent.json:

async with A2AClient("http://localhost:8000") as client:
    card = await client.get_agent_card()
    
    print(card.name)           # "Research Assistant"
    print(card.capabilities)   # AgentCapabilities(streaming=True, ...)
    print(card.skills)         # [AgentSkill(name="search_web", ...)]

Get Extended Card

Fetch detailed agent information (requires authentication):

client = A2AClient("http://localhost:8000", auth_token="sk-secret")
extended = await client.get_extended_card()
print(extended.advanced_features)

---

Message Methods

Send Message

Send a text message and get the complete task result:

# Basic message
result = await client.send_message("Explain quantum computing")

# Access task details
task = result["result"]
print(task["status"]["state"])  # "completed"
print(task["artifacts"])         # Agent response artifacts

Multi-turn conversations with context:

# First message
result1 = await client.send_message("What is Python?")
context_id = result1["result"]["contextId"]

# Follow-up message
result2 = await client.send_message(
    "What about its type system?",
    context_id=context_id
)

Streaming Messages

Send a message and receive real-time Server-Sent Events:

async for event in client.send_message_streaming("Write a haiku"):
    if "result" in event:
        task = event["result"]
        
        # Check task status updates
        if "status" in task:
            print(f"Status: {task['status']['state']}")
        
        # Process response chunks
        if "artifact" in task:
            for part in task["artifact"]["parts"]:
                print(part.get("text", ""))

---

Task Management

Get Task Details

task = await client.get_task("task-uuid-123")
print(task["result"]["status"]["state"])  # "completed", "running", "failed"

List Tasks

# List all tasks
all_tasks = await client.list_tasks()
print(f"Total tasks: {len(all_tasks['result'])}")

# List tasks for specific conversation
ctx_tasks = await client.list_tasks(context_id="ctx-uuid-123")

Cancel Task

cancelled = await client.cancel_task("task-uuid-123")
print(cancelled["result"]["status"]["state"])  # "cancelled"

---

Context Manager Pattern

Recommended approach with automatic session cleanup:

# Auto-closes HTTP session
async with A2AClient("http://localhost:8000") as client:
    card = await client.get_agent_card()
    result = await client.send_message("Hello!")

Manual management when needed:

client = A2AClient("http://localhost:8000")
try:
    result = await client.send_message("Hello!")
finally:
    await client.close()  # Must call explicitly

---

Agent-to-Agent Communication

Complete example showing agent-to-agent delegation:

import asyncio
from praisonaiagents import Agent, A2A, A2AClient

# Start remote agent server (separate process)
# agent = Agent(name="Expert", role="AI Expert", goal="Answer questions")
# a2a = A2A(agent=agent)
# a2a.serve(port=8001)

async def main():
    # Client agent calls remote agent
    async with A2AClient("http://localhost:8001") as remote:
        # Discover capabilities
        card = await remote.get_agent_card()
        print(f"Connected to: {card.name}")
        
        # Delegate task
        result = await remote.send_message("What are the top AI trends?")
        task = result["result"]
        print(f"Status: {task['status']['state']}")
        
        # Stream long responses
        async for event in remote.send_message_streaming("Write a detailed report"):
            if "artifact" in event.get("result", {}):
                for part in event["result"]["artifact"]["parts"]:
                    print(part.get("text", ""), end="")

asyncio.run(main())

---

Error Handling

Handle common HTTP and connection errors:

import aiohttp

try:
    result = await client.send_message("Hello")
    
except aiohttp.ClientResponseError as e:
    if e.status == 401:
        print("Authentication required")
    elif e.status == 404:
        print("A2A endpoint not found")
    elif e.status == 500:
        print("Server error")
        
except aiohttp.ClientError as e:
    print(f"Connection error: {e}")
    
except asyncio.TimeoutError:
    print("Request timed out")

---

Common Patterns

Agent Discovery

Always fetch the agent card first to understand capabilities before sending messages.

async with A2AClient(url) as client:
    card = await client.get_agent_card()
    
    # Check if streaming is supported
    if card.capabilities.streaming:
        # Use streaming for long responses
        async for event in client.send_message_streaming(text):
            process_event(event)
    else:
        # Use regular messaging
        result = await client.send_message(text)

Context Preservation

Maintain conversation context across multiple message exchanges.

context_id = None

async with A2AClient(url) as client:
    # Start conversation
    result1 = await client.send_message("Hello", context_id=context_id)
    context_id = result1["result"]["contextId"]
    
    # Continue conversation  
    result2 = await client.send_message("Follow up", context_id=context_id)

Task Monitoring

Monitor long-running tasks with periodic status checks.

async def monitor_task(client, task_id):
    while True:
        task = await client.get_task(task_id)
        state = task["result"]["status"]["state"]
        
        if state in ["completed", "failed", "cancelled"]:
            break
            
        await asyncio.sleep(1)  # Poll every second
    
    return task

Batch Operations

Process multiple tasks concurrently using asyncio.

async def batch_messages(client, messages):
    tasks = []
    
    for msg in messages:
        task = asyncio.create_task(client.send_message(msg))
        tasks.append(task)
    
    results = await asyncio.gather(*tasks)
    return results

---

Best Practices

Session Management

Always use the context manager pattern to ensure proper session cleanup and avoid connection leaks.

# ✅ Good - automatic cleanup
async with A2AClient(url) as client:
    result = await client.send_message("Hello")

# ❌ Bad - manual cleanup required
client = A2AClient(url)
result = await client.send_message("Hello")
# Session not closed!

Error Recovery

Implement retry logic for transient network errors and timeouts.

import asyncio
from tenacity import retry, stop_after_attempt, wait_exponential

@retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=4, max=10))
async def robust_send_message(client, message):
    try:
        return await client.send_message(message)
    except aiohttp.ClientError as e:
        print(f"Retry after error: {e}")
        raise

Authentication

Store auth tokens securely and never expose them in logs or error messages.

import os

# ✅ Good - from environment
auth_token = os.getenv("A2A_AUTH_TOKEN")
client = A2AClient(url, auth_token=auth_token)

# ❌ Bad - hardcoded token
client = A2AClient(url, auth_token="sk-hardcoded-secret")

Streaming Best Practices

Handle streaming events gracefully with proper error handling and backpressure.

async def process_stream(client, message):
    try:
        async for event in client.send_message_streaming(message):
            if "result" in event:
                # Process incrementally to avoid memory buildup
                process_event_chunk(event["result"])
                
    except asyncio.CancelledError:
        print("Stream cancelled")
        raise
    except Exception as e:
        print(f"Stream error: {e}")
        # Handle gracefully

---

Related

A2A Protocol

Setup A2A servers and protocol basics

Multi-Agent Patterns

Agent orchestration and communication patterns