A2A Client

The A2AClient provides a Python interface for communicating with A2A servers using the implemented HTTP REST or JSON-RPC binding. Match the server binding and endpoint; gRPC is not supported. Examples include:

Agno AgentOS instances with A2A interface enabled
Google ADK agents
Other servers supporting the same REST or JSON-RPC contract

Prerequisites

Install uv pip install -U "agno[os]" for the Python client. Start the Agno A2A interface example, including its server dependencies and model key. It serves my_agent on port 7777. The calls below assume that local server runs without authentication; the Authentication section shows how to pass credentials for a protected deployment.

Quick Start

Connecting to Agno AgentOS via A2A interface

import asyncio
from agno.client.a2a import A2AClient

async def main():
    # Connect to an Agno AgentOS A2A endpoint
    client = A2AClient("http://localhost:7777/a2a/agents/my_agent")

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

asyncio.run(main())

Connecting to Google ADK

This alternative requires a separately running Google ADK A2A server at http://localhost:8001/, configured to serve JSON-RPC. The Agno server above does not create it. See the ADK A2A setup.

import asyncio

from agno.client.a2a import A2AClient

async def main():
    client = A2AClient("http://localhost:8001/", protocol="json-rpc")
    result = await client.send_message(message="Hello!")
    print(result.content)

asyncio.run(main())

Streaming Responses

Stream responses in real-time:

import asyncio

from agno.client.a2a import A2AClient

async def main():
    client = A2AClient("http://localhost:7777/a2a/agents/my_agent")

    async for event in client.stream_message(message="Tell me a story"):
        if event.is_content and event.content:
            print(event.content, end="", flush=True)

asyncio.run(main())

Authentication

AgentOS instances running with authorization=True require a JWT on every A2A request. Pass it via headers:

import asyncio
import os

from agno.client.a2a import A2AClient

async def main():
    client = A2AClient("https://my-agent-os.com/a2a/agents/my-agent")
    headers = {"Authorization": f"Bearer {os.environ['AGENT_OS_JWT']}"}

    await client.send_message(message="Hello!", headers=headers)

    async for event in client.stream_message(message="Hello!", headers=headers):
        ...

asyncio.run(main())

send_message, stream_message, and get_agent_card all accept headers. The token needs the target's run scope (agents:run, or per-resource agents:my-agent:run) for message:send and message:stream. See Scopes for the full mapping.

The user_id parameter sets userId in the message metadata. AgentOS honors it for anonymous callers only. When the request carries a JWT, the run is attributed to the token's principal and user_id is ignored.