> ## Documentation Index
> Fetch the complete documentation index at: https://agno-v2-codex-docs-audit-20260719-0149.mintlify.site/llms.txt
> Use this file to discover all available pages before exploring further.

# A2A Client

> Connect to any A2A-compatible agent server

The `A2AClient` provides a Python interface for communicating with any [A2A protocol](https://a2a-protocol.org/) compatible server. This includes:

* Agno AgentOS instances with A2A interface enabled
* Google ADK agents
* Any other A2A-compatible agent server

## Quick Start

### Connecting to Agno AgentOS via A2A interface

```python theme={null}
import asyncio
from agno.client.a2a import A2AClient

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

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

asyncio.run(main())
```

### Connecting to Google ADK

Google ADK uses JSON-RPC mode:

```python theme={null}
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:

```python theme={null}
import asyncio

from agno.client.a2a import A2AClient


async def main():
    client = A2AClient("http://localhost:7003/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`:

```python theme={null}
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']}"}

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

    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](/agent-os/security/authorization/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.

## Developer Resources

* [A2A Protocol Documentation](https://a2a-protocol.org/)
* [A2AClient Reference](/reference/clients/a2a-client)
* [Agno A2A Interface](/agent-os/interfaces/a2a/introduction)