Agent-to-agent (A2A)

pip install a2a-sdk httpx

import asyncio
import uuid

import httpx
from a2a.client import A2ACardResolver, ClientFactory
from a2a.client.client import ClientConfig
from a2a.types import Message, Part, Role, SendMessageRequest

CARD_URL = "https://ALGOLIA_APPLICATION_ID.algolia.net/agent-studio/1/a2a/apps/ALGOLIA_APPLICATION_ID/agents/AGENT_ID/.well-known/agent-card.json"
API_KEY = "ALGOLIA_SEARCH_API_KEY"


async def main() -> None:
    base_url, _, card_path = CARD_URL.partition("/1/a2a/")
    headers = {"X-Algolia-API-Key": API_KEY}

    async with httpx.AsyncClient(headers=headers, timeout=120) as http:
        resolver = A2ACardResolver(http, base_url, agent_card_path=f"/1/a2a/{card_path}")
        card = await resolver.get_agent_card()
        print("Skills:", [skill.name for skill in card.skills])

        client = ClientFactory(ClientConfig(httpx_client=http)).create(card)
        request = SendMessageRequest(
            message=Message(
                message_id=str(uuid.uuid4()),
                role=Role.ROLE_USER,
                parts=[Part(text="Recommend three classic science fiction movies from the 1980s.")],
            )
        )
        async for response in client.send_message(request):
            if response.HasField("message"):
                print("".join(part.text for part in response.message.parts))


asyncio.run(main())

curl "https://$ALGOLIA_APPLICATION_ID.algolia.net/agent-studio/1/a2a/apps/$ALGOLIA_APPLICATION_ID/agents/$AGENT_ID/rpc" \
  -X POST \
  -H 'Content-Type: application/json' \
  -H 'A2A-Version: 1.0' \
  -H "X-Algolia-API-Key: $ALGOLIA_SEARCH_API_KEY" \
  --data-raw '{
    "jsonrpc": "2.0",
    "id": "1",
    "method": "SendMessage",
    "params": {
      "message": {
        "messageId": "'"$(uuidgen)"'",
        "role": "ROLE_USER",
        "parts": [{ "text": "Recommend three classic science fiction movies from the 1980s." }]
      }
    }
  }'

{
  "jsonrpc": "2.0",
  "id": "1",
  "result": {
    "message": {
      "messageId": "alg_msg_knCtTn8KYJCw1F0H",
      "contextId": "985f1c2a-ca8a-4012-87a2-996d04fa6715",
      "role": "ROLE_AGENT",
      "parts": [{ "text": "Three picks from the 1980s: ..." }]
    }
  }
}

pip install "google-adk[a2a]"

import asyncio

import httpx
from google.adk.agents import LlmAgent
from google.adk.agents.remote_a2a_agent import RemoteA2aAgent
from google.adk.runners import Runner
from google.adk.sessions import InMemorySessionService
from google.genai import types

CARD_URL = "https://ALGOLIA_APPLICATION_ID.algolia.net/agent-studio/1/a2a/apps/ALGOLIA_APPLICATION_ID/agents/AGENT_ID/.well-known/agent-card.json"
API_KEY = "ALGOLIA_SEARCH_API_KEY"


async def main() -> None:
    async with httpx.AsyncClient(headers={"X-Algolia-API-Key": API_KEY}, timeout=120) as http:
        movies_agent = RemoteA2aAgent(
            name="movies_agent",
            description="Recommends movies from the catalog.",
            agent_card=CARD_URL,
            httpx_client=http,
        )
        concierge = LlmAgent(
            name="concierge",
            model="gemini-2.5-flash",
            instruction="Transfer movie questions to movies_agent. Answer other questions yourself.",
            sub_agents=[movies_agent],
        )

        sessions = InMemorySessionService()
        runner = Runner(agent=concierge, app_name="concierge", session_service=sessions)
        session = await sessions.create_session(app_name="concierge", user_id="user-1")
        message = types.Content(role="user", parts=[types.Part(text="Recommend a classic 1980s science fiction movie.")])

        async for event in runner.run_async(user_id="user-1", session_id=session.id, new_message=message):
            if event.content and event.content.parts and event.content.parts[0].text:
                print(f"{event.author}: {event.content.parts[0].text}")


asyncio.run(main())