Build A2A agent mesh

Introduction

When you build multi-agent systems, splitting intelligence across domain specialists avoids the coherence problems that plague single monolithic agents: one model asked to analyze data, draft prose, and validate compliance simultaneously will sacrifice depth in every area. In this lesson you will create three specialist ADK agents—an AnalysisAgent, a WritingAgent, and a ValidationAgent—each backed by a Gemini model, equipped with typed Python tools, and advertised to the mesh via an A2A agent card. By the end, you'll have a working specialist layer that an orchestrator can discover and delegate tasks to with precision.

Key Terminology

• Specialist Agent — an ADK Agent instance scoped to a single capability domain (analysis, writing, or validation) whose instruction string and tool set are deliberately narrow, preventing it from crossing into another agent's responsibility.
• FunctionTool — an ADK wrapper (google.adk.tools.FunctionTool) that registers a typed Python function as a model-callable tool, letting the foundation model invoke compute_statistics, detect_anomalies, render_section, or score_compliance during inference.
• Agent Instruction — the instruction string passed to Agent(...) that constrains the model's behavior to its assigned domain; for example, the writing_agent instruction explicitly forbids data analysis, enforcing role separation at runtime.
• A2A Agent Card — a structured metadata descriptor that advertises a specialist agent's name, capability domain, and endpoint to the mesh so an orchestrator can discover and delegate to it without hardcoded routing logic.
• Responsibility Bleed — the failure mode in monolithic agents where one model handles competing tasks (analysis, prose, compliance) simultaneously and sacrifices depth in each; specialist agents with scoped instructions prevent this by construction.
• Z-score Threshold — the statistical measure used in detect_anomalies to flag outliers: a value's distance from the mean divided by standard deviation, compared against a configurable threshold (default 2.0) to determine whether a data point is anomalous.

Concepts

Why Specialists Beat Monoliths

A single agent asked to analyze data, draft a report, and validate compliance in one pass is not three agents in a trench coat — it is a model being pulled in three directions at once. Foundation models allocate attention and reasoning across the full prompt context; the more competing responsibilities they carry, the shallower each becomes. An instruction that simultaneously demands statistical rigor, structured prose, and rule-by-rule compliance auditing gives the model no clear optimization target.

Specialist agents solve this by fixing the scope at construction time. Each Agent instance in this lesson carries exactly the tools and instruction needed for one domain. The analysis_agent instruction anchors conclusions to specific numbers; the writing_agent instruction explicitly forbids analysis ("do not analyze data"); the validation_agent instruction demands rule references. These constraints are not polite suggestions — the model's behavior at inference time is shaped by the instruction, so scoping is enforced by design rather than convention.

The ADK Tool-Registration Pattern

ADK's tool model is intentionally simple: write a pure Python function with typed parameters, wrap it in FunctionTool, and pass it in the tools= list when constructing an Agent. The ADK runtime generates a schema from the function's type annotations and registers it as a callable the model can invoke during a response. No framework-specific decorators, no separate tool-registry step.

All three specialists in this lesson share this identical structural template: tool functions → FunctionTool wrappers → Agent(name=..., model=..., instruction=..., tools=[...]). The AnalysisAgent is built first precisely because its shape is the pattern the other two replicate with different domain functions. compute_statistics returns a flat dict of descriptive measures; detect_anomalies filters by z-score. render_section produces markdown-headed blocks. score_compliance compares rules against findings and returns a pass-rate dict. Each function is independently testable before ADK registration — calling compute_statistics([10.0, 20.0, 15.0]) directly should return all six expected keys, confirming the tool is correctly typed before the model ever sees it (see Code Walkthrough).

A2A Agent Cards as the Discovery Layer

Building three capable specialist agents is necessary but not sufficient for a mesh — an orchestrator that cannot discover what each agent does will either hardcode routing or fail to delegate at all. A2A agent cards are the solution: each card is a structured descriptor that names the agent, declares its capability domain, and specifies its network endpoint. The orchestrator reads cards rather than source code, which means specialists can be added, replaced, or versioned without touching orchestration logic.

In this lesson the specialists are constructed first; the A2A card attachment happens in the mesh-wiring step that follows. This ordering matters — the agent's name field (e.g., "analysis_agent") and its functional boundary (expressed through its instruction) must be stable before the card is authored, because the card is a contract that downstream orchestrators will depend on.

The card does not duplicate the instruction string — it declares capability categories at a coarser grain ("data analysis", "technical writing", "compliance validation") so the orchestrator can route by domain without parsing prose. The instruction string is the agent's internal contract with the model; the card is the agent's external contract with the mesh.

Code Walkthrough

Now that you understand how ADK wraps a foundation model with typed tools and how A2A agent cards expose each agent's capabilities to the mesh, the implementation follows directly from those two concepts.

The AnalysisAgent is defined first because its structure—tool functions, a FunctionTool wrapper, and an Agent instance with a scoped instruction—is the template all three specialists share. Two pure Python functions become tools: compute_statistics returns descriptive measures, and detect_anomalies flags values whose z-score exceeds a configurable threshold.

Code snippetpython
1import statistics
2from google.adk.agents import Agent
3from google.adk.tools import FunctionTool
4
5def compute_statistics(data_points: list[float]) -> dict:
6    return {
7        "mean": statistics.mean(data_points),
8        "median": statistics.median(data_points),
9        "stdev": statistics.stdev(data_points) if len(data_points) > 1 else 0.0,
10        "min": min(data_points),
11        "max": max(data_points),
12        "count": len(data_points),
13    }
14
15def detect_anomalies(data_points: list[float], threshold: float = 2.0) -> list[dict]:
16    mean = statistics.mean(data_points)
17    stdev = statistics.stdev(data_points) if len(data_points) > 1 else 0.0
18    return [
19        {"index": i, "value": v, "z_score": round(abs(v - mean) / stdev, 3)}
20        for i, v in enumerate(data_points)
21        if stdev > 0 and abs(v - mean) / stdev > threshold
22    ]
23
24analysis_agent = Agent(
25    name="analysis_agent",
26    model="gemini-2.0-flash",
27    instruction="""You are an enterprise data analysis specialist. Analyze datasets,
28compute statistics, and detect anomalies using the provided tools. Always ground
29conclusions in data before forming findings. Report specific numbers, not vague qualifiers.""",
30    tools=[FunctionTool(compute_statistics), FunctionTool(detect_anomalies)],
31)

The WritingAgent and ValidationAgent follow the identical structural pattern with domain-specific tools. The separation of concerns is enforced through each agent's instruction string, which constrains it to its domain and prevents responsibility bleed across agents.

Code snippetpython
1def render_section(title: str, content: str, level: int = 2) -> str:
2    return f"{'#' * level} {title}\n\n{content}"
3
4def score_compliance(rules: list[str], findings: list[str]) -> dict:
5    passed = [r for r in rules if any(r in f for f in findings)]
6    return {"passed": len(passed), "total": len(rules), "score": len(passed) / max(len(rules), 1)}
7
8writing_agent = Agent(
9    name="writing_agent",
10    model="gemini-2.0-flash",
11    instruction="You are an enterprise technical writer. Compose clear, structured reports using the render_section tool. Do not analyze data—only structure and express content you are given.",
12    tools=[FunctionTool(render_section)],
13)
14
15validation_agent = Agent(
16    name="validation_agent",
17    model="gemini-2.0-flash",
18    instruction="You are a compliance validation specialist. Evaluate outputs against business rules using the score_compliance tool. Report pass/fail results with specific rule references.",
19    tools=[FunctionTool(score_compliance)],
20)

Each agent is now an independent, discoverable unit. The A2A agent card for each—describing its name, capability domain, and endpoint—will be attached in the mesh wiring step so the orchestrator can route delegations without hardcoding which agent handles which domain.

Confirm that all three Agent instances initialize without error and that calling compute_statistics([10.0, 20.0, 15.0]) directly returns a dict containing mean, median, stdev, min, max, and count keys, verifying that the tool functions are correctly typed before ADK registers them with the model.

Do's and Don'ts

Having walked through the material above, the following Do's and Don'ts distill it into practice.

Do's

1. ✓Do write each agent's instruction string as an explicit domain contract — the WritingAgent's clause "Do not analyze data—only structure and express content you are given" is the load-bearing mechanism that prevents responsibility bleed; without it, the Gemini model will attempt cross-domain reasoning and undermine the specialist split.
2. ✓Do call your tool functions directly before registering them with FunctionTool — invoking compute_statistics([10.0, 20.0, 15.0]) and checking that all six expected keys (mean, median, stdev, min, max, count) are present confirms correct Python typing before ADK introspects the signature and exposes the function to the model.
3. ✓Do wrap each tool function in its own FunctionTool instance in the tools= list — ADK reads the FunctionTool wrapper to derive the JSON schema it advertises to the model; sharing one wrapper across functions or passing bare callables silently drops the type metadata the model needs to invoke the tool correctly.

Don'ts

1. ✗Don't merge AnalysisAgent, WritingAgent, and ValidationAgent into a single Agent — a monolithic agent asked to run compute_statistics, call render_section, and call score_compliance within one instruction will sacrifice statistical precision, prose clarity, or rule specificity on every complex task, which is exactly the coherence problem the three-specialist architecture exists to prevent.
2. ✗Don't write permissive instruction strings that omit domain boundaries — omitting explicit exclusions (such as the ValidationAgent's "Report pass/fail results with specific rule references") allows the model to fill gaps with uncontrolled inference, producing outputs that look valid but bypass the score_compliance tool entirely.
3. ✗Don't rely on detect_anomalies when stdev is zero — the function guards against division by zero only when stdev > 0; passing a dataset of identical values returns an empty list silently, so callers must check list length before treating an empty result as "no anomalies" rather than "degenerate input."

Implement agent discovery and delegation

Introduction

Engineers often need to route tasks across a fleet of specialized agents without hardcoding which agent handles which capability. When the orchestrator has no runtime view of what downstream agents can do, it either routes blindly or fails silently — both outcomes compound as the fleet grows and agents evolve between deployments. This lesson teaches you to implement an AgentRegistry that discovers agents dynamically by fetching their A2A agent cards, indexes declared skills into a queryable structure, tracks agent health, and exposes lookup methods that a delegation engine can use to route tasks to the right agent at runtime.

Key Terminology

• Agent Card — The JSON document an A2A-compliant agent serves at /.well-known/agent.json, declaring the agent's name, description, and list of skills (each with an id and searchable tags) that the registry parses to learn what the agent can do.
• Skill Index — The skill_index dictionary in AgentRegistry that inverts the agent-card data by mapping each skill tag to the names of healthy agents advertising it, enabling O(1) tag lookups without scanning every registered agent on every routing decision.
• RegisteredAgent — The dataclass that stores complete runtime state for one downstream agent: the raw card JSON, the parsed skills list, a healthy flag, and the last_seen timestamp from the most recent successful card fetch.
• Health Flag — The healthy boolean on RegisteredAgent that is set to False on any failed card refresh and back to True on success; the registry consults this flag before every routing decision, and an unhealthy agent stays registered so it recovers automatically on the next refresh_all call without requiring re-registration.
• Tag-Based Routing — The find_agents_by_tag lookup strategy that returns all healthy agents advertising a given skill tag, supporting pool-style routing where the orchestrator can select among multiple capable agents.
• Skill-ID Routing — The find_agent_by_skill_id lookup strategy that returns the first healthy agent whose card declares an exact skill id, supporting deterministic routing when a precise capability match is required over a pool.

Concepts

Dynamic Discovery Over Hardcoded Routing

When an orchestrator hardcodes which agent handles which task, every fleet change — a new agent, a removed capability, a URL update — requires a code deployment. A runtime registry decouples the orchestrator from the fleet topology: agents self-describe through their cards, the registry discovers and indexes those descriptions, and the orchestrator routes based on what it finds rather than what was true when the code was last written. The AgentRegistry.register call seeds the registry with only a name and URL; actual capability knowledge arrives via the card fetch, not via constructor arguments. This means the fleet can evolve between deployments and the orchestrator adapts on the next refresh cycle without a code change.

The A2A Agent Card as a Capability Contract

The A2A protocol standardizes how agents advertise capabilities: every compliant agent exposes a JSON document at /.well-known/agent.json. That document is the card. It declares the agent's skills, each carrying an id for exact matching and a tags array for categorical grouping. The registry's _refresh_agent fetches this endpoint, parses skills out of the response, and stores them on the RegisteredAgent. This places the source of truth for what an agent can do on the agent itself — not in a central config file — and the registry stays current by re-fetching periodically through refresh_all (see Code Walkthrough).

Two Complementary Data Structures

The registry maintains two structures that serve different access patterns. The agents dictionary is the authoritative record: it holds every registered agent regardless of health, keyed by name, and supports full introspection. The skill_index is a read-optimized projection that inverts the agents → skills relationship into tags → agent-names, so a tag lookup is an O(1) dict read instead of an O(agents × skills) scan across the whole fleet. These structures are never updated independently — _rebuild_skill_index always derives from the authoritative agents dictionary, guaranteeing consistency. Only healthy agents contribute entries to the skill index, which automatically excludes stale or unreachable agents from routing without evicting them from the registry.

Health Tracking Without Eviction

A failed card fetch is typically a transient event — a network blip should not permanently remove an agent from the fleet. The registry handles this by marking an agent unhealthy (agent.healthy = False) on any failure, leaving the RegisteredAgent entry intact. On the next refresh_all cycle, _refresh_agent retries the card fetch: a successful response flips the health flag back to True and triggers a full skill-index rebuild that includes the agent again. Recovery is automatic and re-registration is never required after a temporary outage. This also means the lookup methods — find_agents_by_tag and find_agent_by_skill_id — can use the health flag as a fast gate without any knowledge of why an agent went down or when it came back.

Code Walkthrough

Now that you understand how A2A agent cards advertise capabilities and how the skill-tag contract enables dynamic routing, the following implementation ties those concepts together in a self-contained registry that fetches cards at runtime and keeps the skill index current.

Code snippetpython
1import httpx
2import asyncio
3from dataclasses import dataclass, field
4from datetime import datetime
5from typing import Optional
6
7@dataclass
8class RegisteredAgent:
9    name: str
10    url: str
11    card: Optional[dict] = None
12    last_seen: Optional[datetime] = None
13    healthy: bool = False
14    skills: list[dict] = field(default_factory=list)
15
16class AgentRegistry:
17    def __init__(self, refresh_interval_seconds: int = 30):
18        self.agents: dict[str, RegisteredAgent] = {}
19        self.skill_index: dict[str, list[str]] = {}
20        self.refresh_interval = refresh_interval_seconds
21        self._client = httpx.AsyncClient(timeout=10.0)
22
23    async def register(self, name: str, url: str) -> None:
24        self.agents[name] = RegisteredAgent(name=name, url=url)
25        await self._refresh_agent(name)
26
27    async def _refresh_agent(self, name: str) -> None:
28        agent = self.agents.get(name)
29        if not agent:
30            return
31        try:
32            response = await self._client.get(
33                f"{agent.url}/.well-known/agent.json"
34            )
35            response.raise_for_status()
36            card = response.json()
37            agent.card = card
38            agent.skills = card.get("skills", [])
39            agent.last_seen = datetime.utcnow()
40            agent.healthy = True
41            self._rebuild_skill_index()
42        except (httpx.HTTPError, Exception):
43            agent.healthy = False
44
45    def _rebuild_skill_index(self) -> None:
46        self.skill_index.clear()
47        for name, agent in self.agents.items():
48            if not agent.healthy:
49                continue
50            for skill in agent.skills:
51                for tag in skill.get("tags", []):
52                    if tag not in self.skill_index:
53                        self.skill_index[tag] = []
54                    if name not in self.skill_index[tag]:
55                        self.skill_index[tag].append(name)
56
57    def find_agents_by_tag(self, tag: str) -> list[RegisteredAgent]:
58        agent_names = self.skill_index.get(tag, [])
59        return [self.agents[n] for n in agent_names if self.agents[n].healthy]
60
61    def find_agent_by_skill_id(self, skill_id: str) -> Optional[RegisteredAgent]:
62        for agent in self.agents.values():
63            if not agent.healthy:
64                continue
65            for skill in agent.skills:
66                if skill.get("id") == skill_id:
67                    return agent
68        return None
69
70    async def refresh_all(self) -> dict[str, bool]:
71        results = {}
72        for name in self.agents:
73            await self._refresh_agent(name)
74            results[name] = self.agents[name].healthy
75        return results

RegisteredAgent stores everything the registry knows about a downstream agent: the raw card JSON, a healthy flag, and the list of skills parsed from the card. The health flag is the fast gate the delegation engine checks before routing — an agent that fails its card fetch stays in the registry but is excluded from all skill lookups until it recovers on the next refresh cycle.

AgentRegistry maintains two complementary structures. The agents dictionary is the authoritative record, keyed by name. The skill_index inverts the relationship: it maps each skill tag to the names of healthy agents that advertise it, enabling O(1) tag lookups instead of scanning every agent on every routing decision. The index is cleared and rebuilt after every successful card fetch rather than updated incrementally — for a fleet of tens of agents, a full rebuild is simpler and always consistent.

_refresh_agent fetches the standard /.well-known/agent.json endpoint defined by the A2A protocol. On success it updates the card, records the timestamp, and marks the agent healthy. On any failure — network timeout, HTTP error, or malformed JSON — it marks the agent unhealthy without evicting it, so recovery is automatic on the next refresh_all call without requiring re-registration.

The two lookup methods serve different routing strategies. find_agents_by_tag returns all healthy agents advertising a given tag, useful when the orchestrator wants to select from a pool. find_agent_by_skill_id returns the first healthy agent whose card declares an exact skill ID, useful for deterministic routing when a precise capability match is required.

Verify by registering a mock agent whose card includes the tag "data-analysis", then calling find_agents_by_tag("data-analysis") — it should return a non-empty list; after the mock agent's URL becomes unreachable and refresh_all() runs, the same call should return an empty list.

Do's and Don'ts

Having walked through the material above, the following Do's and Don'ts distill it into practice.

Do's

1. ✓Do fetch agent capabilities at runtime via /.well-known/agent.json — hardcoding which agent handles which capability breaks silently when agents evolve between deployments; fetching the A2A agent card on register() and every refresh_all() cycle means the registry's skill index always reflects what agents currently advertise.
2. ✓Do gate every routing lookup on agent.healthy — both find_agents_by_tag and find_agent_by_skill_id skip unhealthy agents so a card-fetch failure or network timeout never surfaces a dead agent to the delegation engine; the agent stays in self.agents and recovers automatically on the next refresh without requiring re-registration.
3. ✓Do rebuild skill_index fully after each successful card fetch rather than patching it incrementally — a full _rebuild_skill_index() call clears and repopulates the tag-to-agent-names mapping from the current healthy set, keeping the inverted index always consistent with self.agents and enabling O(1) tag lookups at delegation time.

Don'ts

1. ✗Don't route tasks by scanning self.agents on every delegation decision — iterating all agents to match a tag defeats the purpose of skill_index; find_agents_by_tag exists precisely to do an O(1) dict lookup against the pre-built inverted index, and bypassing it reintroduces the linear scan the index was designed to eliminate.
2. ✗Don't evict an agent from the registry when its card fetch fails — marking agent.healthy = False on any httpx.HTTPError or exception is the correct recovery path; removing the RegisteredAgent entry entirely would require the caller to re-register the agent after a transient network blip, losing the last_seen timestamp and breaking automatic recovery on the next refresh_all() cycle.
3. ✗Don't use find_agent_by_skill_id when the orchestrator only has a tag and no exact skill ID — find_agent_by_skill_id iterates every skill on every healthy agent to match a precise skill.get("id") value; calling it with a loose tag string will silently return None because tag matching is handled exclusively by skill_index via find_agents_by_tag.