A2AClient | Agno

The A2AClient provides an async interface for communicating with A2A protocol servers over REST or JSON-RPC. This includes Agno AgentOS instances with A2A enabled and Google ADK agents. Other transports, such as gRPC, are not supported by this client.

Basic Usage

Install uv pip install 'agno[os,a2a]'. Start the A2A introduction server, which exposes my_agent at http://localhost:7777/a2a/agents/my_agent; the server needs its documented model key.

import asyncio

from agno.client.a2a import A2AClient


async def main():
    client = A2AClient("http://localhost:7777/a2a/agents/my_agent")
    result = await client.send_message(message="Hello!")
    print(result.content)


asyncio.run(main())

Later examples are fragments inside an async function like main() above.

Parameters

Parameter	Type	Default	Description
`base_url`	`str`	Required	Base URL of the A2A server. For Agno servers, include the full agent path (e.g., `"http://localhost:7777/a2a/agents/my_agent"`)
`timeout`	`int`	`30`	Request timeout in seconds
`protocol`	`Literal["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:7777/a2a/agents/my_agent")

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

Google ADK

This fragment requires a separately running Google ADK server on port 8001. 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:

Parameter	Type	Default	Description
`message`	`str`	Required	The text message to send
`context_id`	`Optional[str]`	`None`	Context/session ID for multi-turn conversations
`user_id`	`Optional[str]`	`None`	User identifier
`images`	`Optional[List[Image]]`	`None`	Images to include
`audio`	`Optional[List[Audio]]`	`None`	Audio files to include
`videos`	`Optional[List[Video]]`	`None`	Videos to include
`files`	`Optional[List[File]]`	`None`	Files to include
`metadata`	`Optional[Dict[str, Any]]`	`None`	Additional metadata
`headers`	`Optional[Dict[str, str]]`	`None`	HTTP headers to include in the request

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():

Property	Type	Description
`task_id`	`str`	Unique task identifier
`context_id`	`str`	Context/session ID for multi-turn conversations
`status`	`str`	Task status (`"completed"`, `"failed"`, `"canceled"`)
`content`	`str`	Response text content
`artifacts`	`List[Artifact]`	Any artifacts produced (files, images, etc.)
`metadata`	`Optional[Dict]`	Additional response metadata
`is_completed`	`bool`	True if task completed successfully
`is_failed`	`bool`	True if task failed
`is_canceled`	`bool`	True if task was canceled

StreamEvent

Yielded by stream_message():

Property	Type	Description
`event_type`	`str`	Event type (`"content"`, `"reasoning"`, `"working"`, `"completed"`, `"failed"`, `"canceled"`, `"task"`)
`content`	`Optional[str]`	Text content (for content events)
`task_id`	`Optional[str]`	Task identifier
`context_id`	`Optional[str]`	Context/session ID
`metadata`	`Optional[Dict]`	Event metadata
`is_final`	`bool`	True if this is the final event
`is_content`	`bool`	True if this is a content event with text
`is_started`	`bool`	Checks for `started`; the current parser emits `working` instead. Use `event.event_type == "working"` for that state.
`is_completed`	`bool`	True if this is a task completed event
`is_tool_call`	`bool`	Checks legacy `tool_call_started`/`tool_call_completed` types, which the current parser does not emit.

Artifact

Represents files, images, or other artifacts from a task:

Property	Type	Description
`artifact_id`	`str`	Unique artifact identifier
`name`	`Optional[str]`	Artifact name
`description`	`Optional[str]`	Artifact description
`mime_type`	`Optional[str]`	MIME type of the artifact
`uri`	`Optional[str]`	URI to access the artifact
`content`	`Optional[bytes]`	Raw content (if available)

AgentCard

Describes the capabilities of an A2A agent:

Property	Type	Description
`name`	`str`	Agent name
`url`	`str`	Agent URL
`description`	`Optional[str]`	Agent description
`version`	`Optional[str]`	Agent version
`capabilities`	SDK annotation: `List[str]`; actual standard card: mapping	Passed through from the server. AgentOS returns a capability object such as `{"streaming": True}`; do not assume a list.
`metadata`	`Optional[Dict]`	Additional metadata

Multi-Turn Conversations

Reuse context_id to address the same conversation. Remembering prior turns also requires persistence and history on the server: for an Agno agent, configure a database and add_history_to_context=True. The minimal introduction server must be extended with those settings before trying this memory example:

from agno.client.a2a import A2AClient

client = A2AClient("http://localhost:7777/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}")  # Requires server-side persistence and history

Error Handling

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

client = A2AClient("http://localhost:7777/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}")

On this page