Most A2A servers expose a public card at a well-known path. (Spec recommends a well-known URL and describes card contents/capabilities.) (a2a-protocol.org)
import httpx, asyncio
WELL_KNOWN = "/.well-known/agent-card.json" # (spec names vary slightly by version)
BASE_URLS = [
"https://example-agent-1.com",
"https://example-agent-2.io",
"http://localhost:10000",
]
async def fetch_agent_cards():
async with httpx.AsyncClient(timeout=15) as client:
cards = []
for base in BASE_URLS:
url = base.rstrip("/") + WELL_KNOWN
try:
r = await client.get(url, headers={"Accept": "application/json"})
r.raise_for_status()
card = r.json()
# optional: filter by advertised skills/capabilities
if "capabilities" in card:
cards.append((base, card))
except Exception:
pass
return cards
cards = asyncio.run(fetch_agent_cards())
for base, card in cards:
print(base, "→", card.get("name"), "skills:", [s.get("id") for s in card.get("skills", [])])A2A registries let agents register and clients search/filter by skill, name, protocol version, etc. (JSON-RPC primary; REST/GraphQL often secondary). Try the open-source A2A Registry (FastAPI based). (a2a-registry.dev)
JSON-RPC query (search by keyword/skills):
import httpx, asyncio, uuid, json
REGISTRY_RPC_URL = "http://localhost:8000/jsonrpc" # or your hosted registry
async def search_agents(query=None, skills=None, protocol_version=None):
payload = {
"jsonrpc": "2.0",
"method": "search_agents",
"params": {"query": query, "skills": skills, "protocol_version": protocol_version},
"id": str(uuid.uuid4())
}
async with httpx.AsyncClient(timeout=15) as client:
r = await client.post(REGISTRY_RPC_URL, json=payload)
r.raise_for_status()
data = r.json()
return data["result"]
agents = asyncio.run(search_agents(query="weather", skills=["weather_forecast"], protocol_version="0.3.0"))
print(json.dumps(agents, indent=2))You can then pull each agent’s
agent_card.urlfrom the registry result and fetch the full AgentCard to configure your client. (a2a-registry.dev)
If you need strict schema checks (CI gate), validate AgentCards against an A2A JSON schema/profile (community specs exist; versions differ). (a2a.plus)
from jsonschema import validate
AGENT_CARD_SCHEMA = {...} # load from your pinned spec/profile
def validate_card(card: dict):
validate(instance=card, schema=AGENT_CARD_SCHEMA)There’s no single global authority today. Patterns the spec endorses:
- Well-known URIs (direct fetch)
- Registries/Catalogs (enterprise/private, domain-specific, or community)
- Direct configuration (you ship the card/URL) (a2a-protocol.org)
The A2A Registry project is a concrete, working example you can self-host and extend (JSON-RPC + REST, search by skills/version). Treat it as an emerging building block—not “the” canonical web-scale index. (a2a-registry.dev)
- Tool choice (classic function-calling): LLM reads free-text tool descriptions and decides which function to call. Ad-hoc metadata.
- A2A: a standardized AgentCard advertises identity, skills, auth schemes, transports, streaming/polling modes, task semantics, etc.; invocation is JSON-RPC with defined request/response + streaming (SSE) guidance. More contract, less prompt-parsing. (a2a-protocol.org)
Practical win: You can search by skill or capability in a registry, fetch the card, and call the agent via a consistent RPC surface—no bespoke prompt glue per tool. (a2a-registry.dev)
It can, depending on how you implement:
- Bigger envelopes: Cards, capability negotiation, and multi-turn task metadata add some overhead.
- Server-side routing: If the agent itself calls other agents/tools, the server bears some prompt cost (good for client tokens, potentially higher server tokens).
- Streaming/polling: Minimal token impact; it’s about transport, not tokens. Specs define blocking vs polling; streaming uses SSE. (a2a-protocol.org)
Mitigations
- Cache & memoize AgentCards and auth handshakes client-side.
- Use skill filters and short system prompts; avoid re-sending long context each hop.
- Apply response compression (summarize deltas) between agents.
- Prefer server-side routing/planning so the client doesn’t resend giant histories.
- Enforce max context + rerank upstream to keep payloads tight.
The protocol is client→server JSON-RPC. Within one call, the server handles the task (optionally emitting a stream via SSE). However, the server can itself act as a client to other A2A agents (chaining/fan-out). That’s two-way at the system level, but not half-duplex “both agents call each other in the same RPC” from the client’s socket. (Spec: message/send, tasks lifecycle, streaming via SSE.) (a2a-protocol.org)
TL;DR: In a single RPC, it’s request→response (plus stream). Bi-directionality emerges when agents call other agents behind the scenes or via new RPCs.
The ingredients exist: standardized AgentCards, JSON-RPC, streaming, and registries for discovery. You can stitch federated registries (enterprise + public) and apply trust/rbac. That’s a pragmatic path toward an “agent internet.” (a2a-protocol.org)
# discovery.py
import httpx, asyncio, uuid
from typing import Iterable, Dict, Any
WELL_KNOWN_CANDIDATES = [
"/.well-known/agent-card.json", # newer naming
"/.well-known/agent.card.json", # alt naming
"/agent.card.json" # “extended” card (often auth-gated)
]
class A2ADiscovery:
def __init__(self, timeout=10.0, headers=None):
self.timeout = timeout
self.headers = headers or {"Accept": "application/json"}
async def fetch_card(self, base_url: str) -> Dict[str, Any] | None:
async with httpx.AsyncClient(timeout=self.timeout, headers=self.headers) as client:
for path in WELL_KNOWN_CANDIDATES:
try:
r = await client.get(base_url.rstrip("/") + path)
if r.status_code == 200:
return r.json()
except Exception:
pass
return None
async def fetch_many(self, bases: Iterable[str]):
results = []
for b in bases:
card = await self.fetch_card(b)
if card:
results.append((b, card))
return results
class A2ARegistryClient:
def __init__(self, rpc_url: str, timeout=10.0, headers=None):
self.rpc_url = rpc_url
self.timeout = timeout
self.headers = headers or {"Content-Type": "application/json"}
async def search(self, query=None, skills=None, protocol_version=None):
payload = {
"jsonrpc": "2.0",
"method": "search_agents",
"params": {"query": query, "skills": skills, "protocol_version": protocol_version},
"id": str(uuid.uuid4())
}
async with httpx.AsyncClient(timeout=self.timeout, headers=self.headers) as client:
r = await client.post(self.rpc_url, json=payload)
r.raise_for_status()
return r.json().get("result", [])
# Example usage:
# discs = asyncio.run(A2ADiscovery().fetch_many(["http://localhost:10000"]))
# reg = asyncio.run(A2ARegistryClient("http://localhost:8000/jsonrpc").search(skills=["search"]))If you want, I can wire this into your existing notebook: (1) query a registry → (2) pick the best agent by skill → (3) fetch its card → (4) instantiate your client → (5) send a message/stream a response.