The A2A (Agent-to-Agent) protocol is a shared specification that lets independently running agents request work from each other, return results, and stream progress in a structured way. Google began publicly pushing it through 2025, and Antigravity has been building first-class support. In practice you almost always do better with a handful of small specialized agents talking to each other than with one big agent that tries to do everything — and A2A is the standardization of that "talking."
This article walks through implementing A2A communication on Antigravity, starting from a minimal example and building up to a production-ready setup. The structure I am going to share is the one I converged on after rewriting prototypes several times.
Why A2A instead of a single agent
Consolidating everything into one agent feels fast at first. A long enough prompt with enough tools can usually handle most things. But as the workload grows, a few problems compound.
The context window saturates. With one agent owning every responsibility, history, tool definitions, examples, and reference docs all stack up — leaving fewer tokens for the actual reasoning.
Evaluation gets very hard. When something fails, you cannot easily say "the planning step was wrong" or "the verification step missed it." With separate agents, the logs tell you immediately where the breakdown occurred.
Tool authorization drifts. File writes, external API calls, payment actions — all in one mouth — eventually get invoked in unexpected places. Splitting them across agents lets you scope tool access cleanly.
A2A solves these by treating each agent as a service. Each agent advertises a set of capabilities and a receiving endpoint, and they exchange structured request/response messages.
Anatomy of an A2A message
A2A is JSON-based, with a JSON-RPC 2.0 base. A minimal request from a Manager Agent calling a Reviewer Agent looks like this.
{
"jsonrpc": "2.0",
"id": "req-001",
"method": "agent.invoke",
"params": {
"capability": "code_review",
"input": {
"diff": "...",
"language": "typescript"
},
"metadata": {
"trace_id": "trace-abc-123",
"deadline_ms": 30000
}
}
}method is typically agent.invoke or agent.stream. params carries the capability name and the input.
The response shape:
{
"jsonrpc": "2.0",
"id": "req-001",
"result": {
"status": "completed",
"output": {
"issues": [
{"severity": "warning", "line": 42, "message": "..."}
],
"summary": "..."
},
"metadata": {
"tokens_used": 1820,
"elapsed_ms": 8421
}
}
}Pay attention to metadata. Returning trace ID and token usage is strongly recommended by the A2A spec — having or not having those flips your operational visibility entirely.
Defining an agent on Antigravity
To turn an Antigravity agent into an A2A endpoint, you add a capabilities section to its config.
# .antigravity/agents/code-reviewer.yaml
name: code-reviewer
description: "TypeScript / Python code review specialist"
model: gemini-3-flash
a2a:
enabled: true
capabilities:
- id: code_review
description: "Review a code diff"
input_schema:
type: object
properties:
diff: {type: string}
language: {type: string, enum: [typescript, python, go]}
required: [diff, language]
output_schema:
type: object
properties:
issues: {type: array}
summary: {type: string}
auth:
method: api_key
header: X-A2A-Tokena2a.enabled: true switches on the receive endpoint. capabilities advertises what the agent does, with JSON Schema input and output. The schemas matter: a caller passing the wrong shape gets rejected immediately, which makes debugging much faster.
Calling another agent
The caller side uses an A2A client. Here is what it looks like with the Antigravity SDK.
from antigravity.a2a import A2AClient
client = A2AClient(
target_agent="code-reviewer",
auth={"api_key": os.environ["CODE_REVIEWER_TOKEN"]}
)
result = await client.invoke(
capability="code_review",
input={
"diff": diff_content,
"language": "typescript"
},
metadata={
"trace_id": trace_id,
"deadline_ms": 30000
}
)
if result.status == "completed":
issues = result.output["issues"]
# downstream processing
elif result.status == "failed":
logger.error(f"Code review failed: {result.error}")deadline_ms is the parameter most people forget. Without it, if the called agent enters a runaway loop, the caller waits forever. In production, set a sane value on every call — I default to 30 seconds, 60 for genuinely long operations.
Streaming responses
For long-running calls you usually want progress. A2A supports this via agent.stream.
async for chunk in client.stream(
capability="long_analysis",
input=large_dataset
):
if chunk.type == "progress":
print(f"Progress: {chunk.percentage}%")
elif chunk.type == "partial_result":
# surface intermediate output to the UI
update_ui(chunk.data)
elif chunk.type == "completed":
final_result = chunk.data
breakStreaming pays off most for user-facing flows. "Three minutes of silence then a result" feels nothing like "a status update every 30 seconds."
Authentication and scopes
The A2A spec leaves auth implementation to you, but at minimum use API keys or OIDC tokens. The shape I run in production:
auth:
method: oidc
issuer: https://auth.example.com
audience: a2a://code-reviewer
scopes:
- capability:code_review:invoke
- metadata:trace:writePer-capability scopes pay back later. You can cleanly say "external agents may only call code_review, internal agents may also call auto_fix," which becomes important the first time you expose any agent outside your own infrastructure.
Designing error handling
A2A errors follow JSON-RPC error codes. The categories I actually handle in production:
try:
result = await client.invoke(...)
except A2AError as e:
if e.code == -32000:
# Server error: target agent internal failure (retryable)
await retry_with_backoff()
elif e.code == -32602:
# Invalid params: input did not match schema (fail fast)
logger.error(f"Bad input: {e.data}")
raise
elif e.code == -32603:
# Internal error: framework-side bug (page on-call)
notify_oncall(e)
elif e.code == 408:
# Timeout: deadline_ms exceeded
# check whether a partial result came back; retry with longer deadline if not
if e.data.get("partial_result"):
return e.data["partial_result"]
await retry_with_longer_deadline()Treat 408 carefully. A partial result may have been returned, and discarding it just to retry is wasted spend.
Tracing across agents
Multi-agent chains are a tracing nightmare without discipline. Generate a trace_id at the entry point and pass it through every downstream A2A call.
async def handle_user_request(user_input):
trace_id = generate_trace_id()
plan = await planner.invoke(
capability="plan",
input={"goal": user_input},
metadata={"trace_id": trace_id}
)
for step in plan.output["steps"]:
result = await executor.invoke(
capability="execute_step",
input=step,
metadata={"trace_id": trace_id} # carry the same trace_id forward
)Pipe these into OpenTelemetry or any trace backend, and you can see "this user request hit five agents, here is how long each leg took" on a single timeline.
Localizing failures
When agents chain, one failure can cascade. The policy I run with: localize first, escalate only if you must.
The caller decides whether the callee's failure should fail the whole flow. A code-review failure usually should not block a deploy — log and skip.
try:
review_result = await reviewer.invoke(...)
findings = review_result.output["issues"]
except A2AError as e:
logger.warning(f"Review skipped due to: {e}")
findings = [] # continue with empty findingsA planning failure, on the other hand, makes downstream work meaningless — escalate immediately. Write down the policy per agent in a small handbook so on-call doesn't have to think during incidents.
Pre-production checklist
Before going to production with A2A, I confirm:
Input and output schemas are defined and validated in CI. deadline_ms is set on every invoke. Auth uses API key strength or higher. trace_id is propagated through every call in the chain. The escalation policy on errors is written down somewhere on-call can read it.
If those are in place, recovering from a broken interaction between agents becomes a tractable problem rather than a guessing game.
Where to start
Resist the urge to build the whole topology at once. Start by wiring two agents together — Manager calling Reviewer is a fine pair. That alone gives you the message format, auth, errors, and tracing in muscle memory. Add a third and fourth from there, with this checklist next to you.
Multi-agent design is genuinely fun, but without a shared protocol like A2A it slides into chaos quickly. Build the scaffolding first, then move fast on top of it.