Build A2A agent card registry with capability advertisement

Introduction

When you wire up a multi-agent system without a discovery layer, every orchestrator ends up hardcoding peer endpoints — and the first time an agent moves, restarts under a new URL, or rotates credentials, downstream callers break silently in production. Google's A2A protocol addresses this with agent cards, JSON descriptors that advertise an agent's identity, capabilities, endpoint, and authentication requirements so peers can find each other dynamically. By the end of this lesson you'll be able to model agent cards with strict Pydantic validation, build a thread-safe registry that indexes them by capability for O(1) discovery, and apply TTL-based eviction so stale or crashed agents stop appearing in lookup results.

Key Terminology

• Agent Card: A JSON descriptor conforming to the A2A specification that advertises an agent's identity, capabilities, endpoint, and authentication requirements so peers can discover it dynamically.
• Capability Index: A reverse mapping from skill_id to the set of agents offering that skill, enabling O(1) capability-based discovery instead of a full scan of all registered cards.
• TTL Eviction: A staleness model in which each card carries a time-to-live; cards whose registration time plus TTL has elapsed are excluded from discovery and periodically removed by a background sweep.

Concepts

What Is an Agent Card and Why It Matters

An agent card in the A2A protocol serves the same role as a service descriptor in a traditional service mesh, but it carries richer semantic metadata. Where a Kubernetes service object declares a hostname and port, an agent card declares:

• Agent Identity: A globally unique identifier, human-readable name, and organizational namespace that supports cross-organization federation (covered in another goal).
• Capability Advertisement: A structured list of skills the agent offers — each skill tagged with an input schema, output schema, and estimated latency class. This enables capability-based routing rather than name-based routing.
• Authentication Requirements: Whether the agent expects OAuth 2.0 bearer tokens, mutual TLS, API keys, or a combination. Peers consult this before initiating task delegation.
• Endpoint Configuration: The base URL, supported A2A protocol versions, and streaming preferences (SSE, WebSocket, or polling).
• Health and Version Metadata: A heartbeat URL, current version string, and a TTL after which the card should be considered stale.

Key terminology for this section:

• Agent Card: A JSON document conforming to the A2A specification that advertises an agent's identity, capabilities, authentication requirements, and endpoint information to peers in the network.
• Capability Advertisement: The act of publishing structured skill descriptors within an agent card so that orchestrators and peer agents can discover and invoke specific functions without prior knowledge of the agent's implementation.
• Schema Validation: The process of verifying that an agent card conforms to the expected Pydantic model or JSON Schema before it is accepted into the registry, preventing malformed entries from corrupting discovery results.
• Discovery Query: A structured request to the registry that filters agent cards by capability name, input/output type compatibility, minimum trust score, or organizational namespace.

Schema Validation Beyond Pydantic

Pydantic validates the structure of the agent card itself, but production registries also need to validate the input_schema and output_schema dictionaries inside each capability against JSON Schema Draft 2020-12. This matters during task delegation — an orchestrator comparing two candidate agents should be able to verify that the agent's declared input schema is compatible with the payload it intends to send. The jsonschema library's Draft202012Validator.check_schema method serves this purpose. Call it inside a custom Pydantic validator on the AgentCapability model, raising a ValueError with a descriptive message if the schema is malformed. This prevents agents from registering capabilities that claim to accept a certain input but provide a syntactically invalid schema — a class of bug that surfaces as cryptic runtime failures during multi-agent orchestration.

Design Decisions and Trade-offs

Two design choices in this lesson are worth surfacing because they reflect real trade-offs rather than obvious wins. First, the registry holds state in-process behind a single threading.Lock rather than delegating to Redis or another shared store. This keeps the implementation small and removes a network hop from the discovery path, but it means a multi-replica deployment needs an external coordination layer (a shared cache, a gossip protocol, or a control-plane database) before it can scale horizontally. Second, staleness is enforced lazily at discovery time AND swept periodically by a background timer. Filtering stale cards inside discover guarantees consumers never see expired entries even if the sweep is late, while the background evict_stale pass keeps memory bounded under high churn — the cost is a small amount of duplicated work, which is the right trade for correctness over micro-optimization.

Code Walkthrough

Building on the agent-card anatomy above, the walkthrough turns those identity, capability, and TTL fields into a validated model and a thread-safe registry.

First, define a strict data model so malformed cards are rejected at registration time rather than propagating to consumers. The AgentCard model nests an AgentCapability per skill, validates the endpoint as a real URL, constrains trust_score to [0.0, 1.0], and runs a validator that rejects duplicate skill_id values within one card.

Code snippetpython
1from pydantic import BaseModel, Field, HttpUrl, field_validator
2from enum import Enum
3from datetime import datetime
4
5class AuthScheme(str, Enum):
6    MTLS = "mtls"
7    OAUTH2 = "oauth2"
8    API_KEY = "api_key"
9    NONE = "none"
10
11class AgentCapability(BaseModel):
12    skill_id: str = Field(..., min_length=1, max_length=128)
13    name: str = Field(..., min_length=1)
14    input_schema: dict = Field(default_factory=dict)
15    output_schema: dict = Field(default_factory=dict)
16
17class AgentCard(BaseModel):
18    agent_id: str = Field(..., min_length=1, max_length=256)
19    name: str = Field(..., min_length=1)
20    organization: str = Field(default="default")
21    endpoint: HttpUrl
22    auth_scheme: AuthScheme = Field(default=AuthScheme.NONE)
23    capabilities: list[AgentCapability] = Field(..., min_length=1)
24    trust_score: float = Field(default=0.5, ge=0.0, le=1.0)
25    ttl_seconds: int = Field(default=300, ge=30)
26    registered_at: datetime = Field(default_factory=datetime.utcnow)
27
28    @field_validator("capabilities")
29    @classmethod
30    def unique_skill_ids(cls, v):
31        ids = [c.skill_id for c in v]
32        if len(ids) != len(set(ids)):
33            raise ValueError("Duplicate skill_id in capabilities")
34        return v

With the model locked down, the registry keeps two structures: a primary agent_id → AgentCard map and a reverse capability index (skill_id → set of agent_id). The index makes capability lookups O(1) instead of a full scan, while discover skips any card past its TTL and evict_stale removes expired entries from both structures under a lock.

Code snippetpython
1from datetime import datetime, timedelta
2from threading import Lock
3
4class A2AAgentCardRegistry:
5    def __init__(self):
6        self._cards: dict[str, AgentCard] = {}
7        self._skill_index: dict[str, set[str]] = {}
8        self._lock = Lock()
9
10    def register(self, card: AgentCard) -> bool:
11        with self._lock:
12            self._cards[card.agent_id] = card
13            for cap in card.capabilities:
14                self._skill_index.setdefault(cap.skill_id, set()).add(card.agent_id)
15        return True
16
17    def _is_stale(self, card: AgentCard) -> bool:
18        expiry = card.registered_at + timedelta(seconds=card.ttl_seconds)
19        return datetime.utcnow() > expiry
20
21    def discover(self, skill_id=None, min_trust=0.0) -> list[AgentCard]:
22        with self._lock:
23            ids = self._skill_index.get(skill_id, set()) if skill_id else set(self._cards)
24            return [
25                c for aid in ids
26                if (c := self._cards.get(aid))
27                and not self._is_stale(c)
28                and c.trust_score >= min_trust
29            ]
30
31    def evict_stale(self) -> int:
32        with self._lock:
33            stale = [aid for aid, c in self._cards.items() if self._is_stale(c)]
34            for aid in stale:
35                card = self._cards.pop(aid)
36                for cap in card.capabilities:
37                    self._skill_index.get(cap.skill_id, set()).discard(aid)
38            return len(stale)

Verify by registering two cards that share a skill_id, calling discover(skill_id=...), and confirming both are returned — then construct a card whose registered_at is older than its ttl_seconds, register it, and confirm evict_stale() returns a non-zero count and that the card no longer appears in discovery results.

Do's and Don'ts

Do's

1. ✓Do declare _skill_index values as set[str], not list[str] — re-calling register() for an already-known agent must be idempotent; a list accumulates duplicate agent_id entries so discover(skill_id=...) returns the same AgentCard multiple times for a single capability query.
2. ✓Do update both _cards and _skill_index inside the same Lock acquisition on every write path — splitting eviction or registration across two separate with self._lock blocks lets a concurrent discover() read a window where the primary map and reverse capability index are temporarily inconsistent, serving ghost or missing results.
3. ✓Do keep the inline _is_stale(c) check inside discover() even when evict_stale() runs on a schedule — between eviction sweeps an agent can cross its registered_at + ttl_seconds expiry boundary, and without the per-call freshness gate, discover() continues serving expired AgentCard entries to callers until the next sweep fires.

Don'ts

1. ✗Don't skip the unique_skill_ids field_validator on AgentCard — if two AgentCapability entries share the same skill_id, _skill_index.setdefault(...).add(agent_id) silently collapses both capabilities under one key, making the duplicate skill effectively undiscoverable by any peer querying that skill_id.
2. ✗Don't remove an agent from _cards in evict_stale() without also purging its skill_id entries from _skill_index — orphaned IDs in the reverse index cause discover() to attempt _cards.get(aid) on a removed agent; the walrus-operator guard drops None silently rather than raising an error, masking the memory leak indefinitely.
3. ✗Don't re-register an agent that has updated its capability set without first clearing its previous skill_id entries from _skill_index — register() only adds to the index and never removes stale entries, so a re-registering agent whose skills changed continues to appear in discover(skill_id=...) results for capabilities it no longer advertises.