Skip to main content
The A2AClient provides an async interface for communicating with any A2A protocol compatible server. This includes Agno AgentOS instances with A2A interface enabled, Google ADK agents, and any other A2A-compatible agent server.

​
Basic Usage

from agno.client.a2a import A2AClient

# Connect to an Agno AgentOS A2A endpoint
client = A2AClient("http://localhost:7001/a2a/agents/my-agent")

# Send a message
result = await client.send_message(message="Hello!")
print(result.content)

​
Parameters

ParameterTypeDefaultDescription
base_urlstrRequiredBase URL of the A2A server. For Agno servers, include the full agent path (e.g., "http://localhost:7003/a2a/agents/my-agent")
timeoutint30Request timeout in seconds
protocolLiteral["rest", "json-rpc"]"rest"Protocol mode. Use "json-rpc" for Google ADK servers

​
Connecting to Different Servers

​
Agno AgentOS

For Agno servers with A2A interface enabled, include the full agent path in the URL: 
from agno.client.a2a import A2AClient

# The URL includes the A2A path to the specific agent
client = A2AClient("http://localhost:7001/a2a/agents/my-agent")

result = await client.send_message(message="What can you help me with?")
print(result.content)

​
Google ADK

Google ADK uses pure JSON-RPC mode. Set protocol="json-rpc": 
from agno.client.a2a import A2AClient

# Google ADK uses JSON-RPC at the root endpoint
client = A2AClient("http://localhost:8001/", protocol="json-rpc")

result = await client.send_message(message="Tell me an interesting fact")
print(result.content)

​
Methods

​
send_message

Send a message to an A2A agent and wait for the complete response. 
result = await client.send_message(
    message="What is the capital of France?",
    user_id="user-123",
    context_id="session-456",
)
print(result.content)
print(f"Task ID: {result.task_id}")
print(f"Context ID: {result.context_id}")
Parameters:
ParameterTypeDefaultDescription
messagestrRequiredThe text message to send
context_idOptional[str]NoneContext/session ID for multi-turn conversations
user_idOptional[str]NoneUser identifier
imagesOptional[List[Image]]NoneImages to include
audioOptional[List[Audio]]NoneAudio files to include
videosOptional[List[Video]]NoneVideos to include
filesOptional[List[File]]NoneFiles to include
metadataOptional[Dict[str, Any]]NoneAdditional metadata
Returns: TaskResult Raises:
  • HTTPStatusError: If the server returns an HTTP error (4xx, 5xx)
  • RemoteServerUnavailableError: If connection fails or times out

​
stream_message

Stream a message response in real-time. 
async for event in client.stream_message(
    message="Tell me a story",
    user_id="user-123",
):
    if event.is_content and event.content:
        print(event.content, end="", flush=True)

    if event.is_final:
        print("\n--- Stream complete ---")
Parameters: Same as send_message Yields: StreamEvent Raises:
  • HTTPStatusError: If the server returns an HTTP error (4xx, 5xx)
  • RemoteServerUnavailableError: If connection fails or times out

​
get_agent_card

Get the agent card for capability discovery. 
card = client.get_agent_card()
if card:
    print(f"Agent: {card.name}")
    print(f"Description: {card.description}")
    print(f"Capabilities: {card.capabilities}")
Returns: AgentCard if available, None otherwise

​
aget_agent_card

Get the agent card for capability discovery asynchronously. 
card = await client.aget_agent_card()
if card:
    print(f"Agent: {card.name}")
    print(f"Description: {card.description}")
    print(f"Capabilities: {card.capabilities}")
Returns: AgentCard if available, None otherwise

​
Response Types

​
TaskResult

Returned by send_message():
PropertyTypeDescription
task_idstrUnique task identifier
context_idstrContext/session ID for multi-turn conversations
statusstrTask status ("completed", "failed", "canceled")
contentstrResponse text content
artifactsList[Artifact]Any artifacts produced (files, images, etc.)
metadataOptional[Dict]Additional response metadata
is_completedboolTrue if task completed successfully
is_failedboolTrue if task failed
is_canceledboolTrue if task was canceled

​
StreamEvent

Yielded by stream_message():
PropertyTypeDescription
event_typestrEvent type ("content", "started", "completed", "failed", "working")
contentOptional[str]Text content (for content events)
task_idOptional[str]Task identifier
context_idOptional[str]Context/session ID
metadataOptional[Dict]Event metadata
is_finalboolTrue if this is the final event
is_contentboolTrue if this is a content event with text
is_startedboolTrue if this is a task started event
is_completedboolTrue if this is a task completed event
is_tool_callboolTrue if this is a tool call event

​
Artifact

Represents files, images, or other artifacts from a task:
PropertyTypeDescription
artifact_idstrUnique artifact identifier
nameOptional[str]Artifact name
descriptionOptional[str]Artifact description
mime_typeOptional[str]MIME type of the artifact
uriOptional[str]URI to access the artifact
contentOptional[bytes]Raw content (if available)

​
AgentCard

Describes the capabilities of an A2A agent:
PropertyTypeDescription
namestrAgent name
urlstrAgent URL
descriptionOptional[str]Agent description
versionOptional[str]Agent version
capabilitiesList[str]List of agent capabilities
metadataOptional[Dict]Additional metadata

​
Multi-Turn Conversations

Use context_id to maintain conversation context across multiple messages: 
from agno.client.a2a import A2AClient

client = A2AClient("http://localhost:7003/a2a/agents/my-agent")

# First message - no context_id
result1 = await client.send_message(
    message="My name is Alice and I love Python.",
)
print(f"Agent: {result1.content}")

# Get context_id from response
context_id = result1.context_id

# Follow-up message - include context_id
result2 = await client.send_message(
    message="What is my name?",
    context_id=context_id,
)
print(f"Agent: {result2.content}")  # Should remember "Alice"

​
Error Handling

from agno.client.a2a import A2AClient
from agno.exceptions import RemoteServerUnavailableError
from httpx import HTTPStatusError

client = A2AClient("http://localhost:7003/a2a/agents/my-agent")

try:
    result = await client.send_message(message="Hello")
except RemoteServerUnavailableError as e:
    print(f"Server unavailable: {e.message}")
    print(f"URL: {e.base_url}")
except HTTPStatusError as e:
    print(f"HTTP error: {e.response.status_code}")