Implementing Antigravity's A2A Protocol — Practical Patterns for Agent-to-Agent Conversation

Any serious multi-agent discussion eventually lands on the same question: how do agents actually talk to each other? Antigravity ships an A2A (Agent-to-Agent) protocol for exactly that — a typed way for one agent to delegate work to another and receive results.

The docs cover the surface, but rarely the "when would I reach for this?" part. This article walks through three patterns I use in practice, each with working code.

A2A in One Minute

A2A treats each agent as an HTTP service that exposes a handful of structured operations. Three concepts do the heavy lifting:

• Agent Card: a self-describing schema for what the agent can do (endpoint, supported task kinds, auth requirements)
• Task: an object the client-side agent creates and hands off, carrying the input message and metadata
• Message Stream: the sequence of messages exchanged while a task is in flight — progress updates, clarifying questions, the final result

In diagram form: a client agent, a remote agent, and the task that mediates between them.

Minimal Setup — Delegating One Task

Agent A hands a long article to Agent B for summarization, and Agent B returns three bullet points.

# agent_b_server.py — the summarizer
from antigravity.a2a import AgentServer, AgentCard, TaskHandler
 
card = AgentCard(
    name="summarizer",
    description="Summarize long-form text in 3 bullet points.",
    supported_tasks=["text/summarize"],
    endpoint="http://localhost:8081",
)
 
handler = TaskHandler(card=card)
 
@handler.on_task("text/summarize")
async def summarize(task):
    text = task.input["text"]
    result = await task.call_llm(
        prompt=f"Summarize this text in 3 bullet points:\n\n{text}",
    )
    return {"summary": result}
 
if __name__ == "__main__":
    AgentServer(handler).run(port=8081)

# agent_a_client.py — the caller
from antigravity.a2a import AgentClient
 
async def main():
    client = AgentClient(remote="http://localhost:8081")
    result = await client.submit_task(
        kind="text/summarize",
        input={"text": open("article.md").read()},
    )
    print(result["summary"])

Two things matter here. The Agent Card declares capabilities in a way the client can discover before connecting. The task kind (text/summarize) is the agreement between the two sides — as long as both use the same string, they're compatible, even if one side swaps its implementation later.

Pattern 1: Fire-and-Forget

The setup above. A throws a task to B, B returns a result, A moves on. This is the most common shape and where you should start.

It fits when:

• Input and output are well-defined (summarize, translate, classify, reshape)
• No human confirmation is needed mid-task
• The result lets A complete its own work

This looks a lot like a REST call, and at one level it is. The real value of A2A here is that agents become replaceable components. Swap the summarizer for a different model, a language-specific variant, or a free tier — the client code doesn't change.

Pattern 2: Bidirectional Confirmation

Sometimes B wants to check in with A mid-task. Maybe B is drafting a document and wants A's approval on a proposed title. A2A's Message Stream makes that two-way conversation first-class.

@handler.on_task("doc/generate")
async def generate_doc(task):
    draft_title = await task.call_llm(prompt="Propose a title for this report: ...")
    
    # Ask A to confirm the title
    response = await task.ask(
        message=f"Is this title OK? '{draft_title}'. Reply 'yes' or a new title.",
    )
    
    final_title = draft_title if response.strip().lower() == "yes" else response.strip()
    # ... generate the body ...
    return {"title": final_title, "body": body}

The pattern suits near-autonomous flows that deliberately pause at a few important decision points. Depending on A's implementation, the confirmation prompt can escalate further — to a senior agent with approval authority, or to a human.

Pattern 3: Scatter-Gather

A fires tasks at B1, B2, B3 in parallel and aggregates the results. Useful whenever you want to saturate throughput.

import asyncio
 
async def process_articles(article_list):
    clients = [AgentClient(remote=url) for url in worker_endpoints]
    tasks = [
        clients[i % len(clients)].submit_task(
            kind="text/summarize",
            input={"text": text},
        )
        for i, text in enumerate(article_list)
    ]
    results = await asyncio.gather(*tasks)
    return [r["summary"] for r in results]

Why not just a queue and workers? Because A2A workers can specialize. A Japanese-focused summarizer, an English one, a code-specific one — route each article to the best fit when you construct the task list, and you get more than just parallelism.

Three Operational Gotchas

Auth and access control. Exposing an A2A endpoint on the open internet lets anyone submit tasks. Declare auth requirements on the Agent Card and enforce them with bearer tokens or mTLS. Even inside a private network, a minimum of token auth is worth it if anything reaches across a VPN.

Timeouts and retries. Remote agents sometimes don't respond. Set a client-side timeout. And if the task can be safely retried, tag it with an idempotency_key so double-submission doesn't double-side-effect.

Error granularity. Letting internal exceptions escape through the wire leaks your implementation. Wrap errors in a structured form — TaskError(code="llm_timeout", message="...", retry_after=30) — so the caller can make a meaningful fallback decision.

Where to Start

All the samples here run by installing Antigravity's Python SDK and starting two terminals. Get fire-and-forget working first, then layer on the bidirectional and scatter-gather patterns once the basics feel natural.

When you're ready to design larger topologies, our intro to multi-agent orchestration covers the trade-offs between A2A, shared memory, and orchestrator-worker shapes.