Agent Discovery with A2A: How Peers Find Each Other

When agents need to collaborate, they first need to find each other. The Agent-to-Agent (A2A) protocol provides a standardized way for agents to publish their capabilities, discover peers, and communicate across organizational boundaries. Think of it as DNS for agents.

Why Agent Discovery Matters

In a single-organization setup, you can hardcode agent endpoints. But in the A2A economy:

• Agents are distributed across multiple organizations
• Capabilities change as agents are updated
• New agents come online dynamically
• Trust relationships vary between peers

Discovery solves these problems by providing a dynamic, standards-based way for agents to find and connect with each other.

The A2A Discovery Model

Agent Cards

Every agent publishes an Agent Card — a machine-readable description of its capabilities:

{
  "name": "Research Assistant",
  "description": "A research agent that finds and summarizes academic papers",
  "url": "https://research.example.com/a2a",
  "version": "1.0.0",
  "capabilities": {
    "streaming": true,
    "pushNotifications": false,
    "stateTransitionHistory": true
  },
  "skills": [
    {
      "name": "paper_search",
      "description": "Search for academic papers by topic, author, or keyword",
      "inputModes": ["text"],
      "outputModes": ["text", "application/json"]
    },
    {
      "name": "paper_summary",
      "description": "Generate a summary of an academic paper",
      "inputModes": ["text", "application/pdf"],
      "outputModes": ["text"]
    },
    {
      "name": "citation_analysis",
      "description": "Analyze citation networks and impact metrics",
      "inputModes": ["text"],
      "outputModes": ["application/json"]
    }
  ],
  "authentication": {
    "schemes": ["bearer"]
  }
}

Discovery Server

A central registry where agents publish their cards:

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel

app = FastAPI()

class AgentCard(BaseModel):
    name: str
    description: str
    url: str
    version: str
    capabilities: dict
    skills: list[dict]
    authentication: dict

class DiscoveryServer:
    def __init__(self):
        self.agents: dict[str, AgentCard] = {}
    
    def register(self, agent_id: str, card: AgentCard):
        """Register an agent's capabilities."""
        self.agents[agent_id] = card
    
    def discover(self, skill: str = None, keyword: str = None) -> list[dict]:
        """Find agents matching criteria."""
        results = []
        
        for agent_id, card in self.agents.items():
            match = True
            
            if skill:
                match = match and any(
                    s["name"] == skill for s in card.skills
                )
            
            if keyword:
                match = match and (
                    keyword.lower() in card.description.lower() or
                    keyword.lower() in card.name.lower()
                )
            
            if match:
                results.append({
                    "agent_id": agent_id,
                    "card": card.dict()
                })
        
        return results

discovery = DiscoveryServer()

@app.post("/register/{agent_id}")
async def register_agent(agent_id: str, card: AgentCard):
    discovery.register(agent_id, card)
    return {"status": "registered", "agent_id": agent_id}

@app.get("/discover")
async def discover_agents(skill: str = None, keyword: str = None):
    return {"agents": discovery.discover(skill, keyword)}

Implementing Agent Discovery

Agent Registration

import httpx
from datetime import datetime

class A2AAgent:
    def __init__(self, config: dict):
        self.config = config
        self.card = self._build_card()
    
    def _build_card(self) -> dict:
        return {
            "name": self.config["name"],
            "description": self.config["description"],
            "url": self.config["a2a_endpoint"],
            "version": self.config["version"],
            "capabilities": {
                "streaming": self.config.get("streaming", True),
                "pushNotifications": self.config.get("push_notifications", False),
                "stateTransitionHistory": True,
            },
            "skills": self.config["skills"],
            "authentication": {
                "schemes": ["bearer"]
            }
        }
    
    async def register_with_discovery(self, discovery_url: str):
        """Register this agent with the discovery server."""
        async with httpx.AsyncClient() as client:
            response = await client.post(
                f"{discovery_url}/register/{self.config['agent_id']}",
                json=self.card
            )
            return response.json()
    
    async def discover_peer(
        self,
        discovery_url: str,
        skill: str = None
    ) -> list[dict]:
        """Find agents with specific capabilities."""
        params = {}
        if skill:
            params["skill"] = skill
        
        async with httpx.AsyncClient() as client:
            response = await client.get(
                f"{discovery_url}/discover",
                params=params
            )
            return response.json()["agents"]

Peer-to-Peer Communication

class A2AClient:
    def __init__(self):
        self.session = httpx.AsyncClient()
    
    async def send_task(
        self,
        agent_url: str,
        task: dict,
        auth_token: str
    ) -> dict:
        """Send a task to a discovered agent."""
        headers = {"Authorization": f"Bearer {auth_token}"}
        
        response = await self.session.post(
            f"{agent_url}/tasks/send",
            json={
                "jsonrpc": "2.0",
                "method": "tasks/send",
                "params": {
                    "id": str(uuid.uuid4()),
                    "message": {
                        "role": "user",
                        "parts": [
                            {"type": "text", "text": task["instruction"]}
                        ]
                    }
                }
            },
            headers=headers
        )
        
        return response.json()
    
    async def stream_task(
        self,
        agent_url: str,
        task: dict,
        auth_token: str
    ):
        """Stream results from a discovered agent."""
        headers = {"Authorization": f"Bearer {auth_token}"}
        
        async with self.session.stream(
            "POST",
            f"{agent_url}/tasks/sendSubscribe",
            json={
                "jsonrpc": "2.0",
                "method": "tasks/sendSubscribe",
                "params": {
                    "id": str(uuid.uuid4()),
                    "message": {
                        "role": "user",
                        "parts": [
                            {"type": "text", "text": task["instruction"]}
                        ]
                    }
                }
            },
            headers=headers
        ) as response:
            async for line in response.aiter_lines():
                if line.startswith("data: "):
                    yield json.loads(line[6:])

Multi-Agent Collaboration Pattern

class AgentCollaborator:
    def __init__(self, discovery_url: str, auth_manager):
        self.discovery_url = discovery_url
        self.auth = auth_manager
        self.client = A2AClient()
    
    async def collaborate(
        self,
        task: str,
        required_skills: list[str]
    ) -> dict:
        """Find agents and collaborate on a task."""
        # Discover agents with required skills
        available_agents = []
        
        for skill in required_skills:
            agents = await self._find_agents_with_skill(skill)
            available_agents.extend(agents)
        
        # Remove duplicates
        unique_agents = {a["agent_id"]: a for a in available_agents}
        
        # Get auth tokens for each agent
        tasks = []
        for agent_id, agent_info in unique_agents.items():
            token = await self.auth.get_token(agent_id)
            
            task = {
                "agent_id": agent_id,
                "agent_url": agent_info["card"]["url"],
                "auth_token": token,
                "instruction": f"Help with this task: {task}",
            }
            tasks.append(task)
        
        # Execute tasks in parallel
        results = await asyncio.gather(*[
            self.client.send_task(
                t["agent_url"],
                {"instruction": t["instruction"]},
                t["auth_token"]
            )
            for t in tasks
        ])
        
        # Synthesize results
        return self._synthesize_results(results)
    
    async def _find_agents_with_skill(self, skill: str) -> list[dict]:
        async with httpx.AsyncClient() as client:
            response = await client.get(
                f"{self.discovery_url}/discover",
                params={"skill": skill}
            )
            return response.json()["agents"]
    
    def _synthesize_results(self, results: list[dict]) -> dict:
        # Combine results from multiple agents
        combined = {
            "agent_results": results,
            "synthesis": self._merge_findings(results)
        }
        return combined

Trust and Authentication

class TrustManager:
    def __init__(self):
        self.trusted_agents: dict[str, dict] = {}
        self.revoked_agents: set[str] = set()
    
    def trust_agent(self, agent_id: str, public_key: str, metadata: dict):
        """Register a trusted agent."""
        self.trusted_agents[agent_id] = {
            "public_key": public_key,
            "metadata": metadata,
            "trusted_at": datetime.now().isoformat()
        }
    
    def is_trusted(self, agent_id: str) -> bool:
        """Check if an agent is trusted."""
        return (
            agent_id in self.trusted_agents and
            agent_id not in self.revoked_agents
        )
    
    def verify_signature(self, agent_id: str, message: str, signature: str) -> bool:
        """Verify an agent's message signature."""
        if not self.is_trusted(agent_id):
            return False
        
        agent = self.trusted_agents[agent_id]
        # Verify using the agent's public key
        return verify_signature(agent["public_key"], message, signature)

Fault Tolerance

class ResilientDiscovery:
    def __init__(self, discovery_servers: list[str]):
        self.servers = discovery_servers
        self.cache = {}
    
    async def discover_with_fallback(
        self,
        skill: str = None
    ) -> list[dict]:
        """Try multiple discovery servers."""
        for server in self.servers:
            try:
                result = await self._query_server(server, skill)
                self.cache[skill or "all"] = {
                    "results": result,
                    "timestamp": datetime.now()
                }
                return result
            except Exception as e:
                logger.warning(f"Discovery server {server} failed: {e}")
                continue
        
        # Use cache if all servers fail
        cached = self.cache.get(skill or "all")
        if cached and (datetime.now() - cached["timestamp"]).seconds < 300:
            logger.info("Using cached discovery results")
            return cached["results"]
        
        raise Exception("All discovery servers unavailable")

Conclusion

Agent discovery via A2A transforms isolated agents into a collaborative ecosystem. By publishing Agent Cards, agents declare their capabilities. By querying discovery servers, agents find the right peers. Combined with authentication and fault tolerance, this enables robust cross-organization agent collaboration. Start with a simple discovery server, add trust management as you scale, and always cache results for resilience.