A2A Client

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

from agno.client.a2a import A2AClient

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

from agno.client.a2a import A2AClient

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)

import os

from agno.client.a2a import A2AClient

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)

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