A2A Protocol: The Agent-to-Agent Communication Standard

Chapter 40: A2A Protocol: The Agent-to-Agent Communication Standard

When a single Agent lacks the capability to complete a complex task, collaboration between specialized Agents becomes inevitable. But how do Agents "talk" to each other? How does one delegate a task? How are intermediate results handed off? This is exactly the problem the A2A (Agent-to-Agent) protocol solves. A2A is a multi-vendor Agent communication standard led by Google, complementing Anthropic's MCP: MCP handles "tool invocation," while A2A handles "Agent delegation." This chapter dissects the A2A protocol and demonstrates its application through a complete case study of two Hermes instances collaborating.

40.1 A2A Background and Design Philosophy

Why A2A?

What MCP solved:
    Agent ──MCP──→ Tool (database, file, API)
    ✓ Tools are stateless "function calls"
    ✓ Tools execute quickly (milliseconds to seconds)
    ✓ Tools don't need to "think"

What MCP didn't solve:
    Agent A ──?──→ Agent B (another AI Agent)
    ✗ Agent B may need minutes to complete its task
    ✗ Agent B has its own reasoning process (not a simple function)
    ✗ Agent B may further delegate to Agent C
    ✗ Agent A cannot discover Agent B's capability boundaries

What A2A solves:
    "How can AI Agents from different vendors and frameworks
     understand each other and delegate tasks"

Google's Design Principles

A2A was proposed by Google in early 2025 with these core principles:

MCP and A2A: A Complementary Relationship

┌──────────────────────────────────────────────────────┐
│              Agent Capability Call Hierarchy          │
├──────────────────────────────────────────────────────┤
│                                                      │
│    User Request                                      │
│         ↓                                           │
│    Hermes Agent (Orchestrator)                       │
│         │                                           │
│         ├── MCP ──→ Tool Layer                       │
│         │          (database, search, filesystem)   │
│         │          Synchronous, stateless, fast     │
│         │                                           │
│         └── A2A ──→ Agent Layer                      │
│                    (specialist agents, sub-agents)  │
│                    Asynchronous, stateful, long-running │
│                                                      │
└──────────────────────────────────────────────────────┘

MCP (tool invocation):
  "I need to query the database" → query_db() → result
  Analogy: calling a function

A2A (Agent delegation):
  "I need a legal compliance analysis" → delegate to Legal Agent →
  wait (may take 5 minutes) → receive complete analysis report
  Analogy: outsourcing to a specialist

40.2 Core Concepts: Task, Artifact, Message

Task

A Task is the fundamental unit of work in A2A—it represents work to be completed by the target Agent.

{
  "id": "task-8f3a92b1-4c2d-4e5f-a123-456789abcdef",
  "status": {
    "state": "working",
    "timestamp": "2025-01-15T09:30:00Z",
    "progress": 0.45
  },
  "sessionId": "session-xxx",
  "metadata": {
    "priority": "high",
    "deadline": "2025-01-15T10:00:00Z"
  }
}

Task state machine:

submitted → working → completed
               ↓           ↓
             failed    (returns Artifacts)
               ↓
            canceled

Artifact

An Artifact is the output produced after Task completion—text, files, or structured data.

{
  "name": "compliance_report",
  "description": "Legal compliance analysis report",
  "parts": [
    {
      "type": "text",
      "text": "## Compliance Analysis Report\n\nBased on the contract provided...",
      "metadata": {"language": "en", "word_count": 2847}
    },
    {
      "type": "file",
      "file": {
        "name": "compliance_report.pdf",
        "mimeType": "application/pdf",
        "bytes": "JVBERi0xLjQK..."
      }
    }
  ]
}

Message

A Message is the communication unit exchanged between Client and Agent, supporting multimedia content.

{
  "role": "user",
  "parts": [
    {
      "type": "text",
      "text": "Please analyze the compliance risks in this contract"
    },
    {
      "type": "file",
      "file": {
        "name": "contract.pdf",
        "mimeType": "application/pdf",
        "uri": "https://storage.example.com/contracts/2025/001.pdf"
      }
    }
  ]
}

40.3 Agent Card Specification

The Agent Card is A2A's most important discovery mechanism—every Agent publishes its capability declaration at the fixed address /.well-known/agent.json.

Complete Agent Card Example

{
  "$schema": "https://a2aprotocol.ai/schema/agent-card.json",
  "name": "Legal Compliance Analyzer",
  "description": "Specializes in commercial contract legal compliance analysis covering Chinese, US, and EU regulations. Identifies potential legal risks and provides amendment suggestions.",
  "version": "2.1.0",
  "url": "https://legal-agent.example.com",
  
  "provider": {
    "organization": "LegalTech Corp",
    "url": "https://legaltech.example.com",
    "contact": "[email protected]"
  },
  
  "capabilities": {
    "streaming": true,
    "pushNotifications": true,
    "stateTransitionHistory": true
  },
  
  "skills": [
    {
      "id": "contract-compliance",
      "name": "Contract Compliance Analysis",
      "description": "Analyze commercial contracts for legal compliance, identifying risky clauses",
      "tags": ["legal", "compliance", "contract", "risk-analysis"],
      "examples": [
        "Analyze whether this procurement contract meets China Contract Law requirements",
        "Identify unequal terms in this employment contract"
      ],
      "inputModes": ["text", "file"],
      "outputModes": ["text", "file"]
    },
    {
      "id": "gdpr-check",
      "name": "GDPR Compliance Check",
      "description": "Check if data processing agreements comply with EU GDPR requirements",
      "tags": ["gdpr", "data-privacy", "eu-compliance"],
      "inputModes": ["text"],
      "outputModes": ["text"]
    }
  ],
  
  "authentication": {
    "schemes": ["bearer", "oauth2"],
    "oauth2": {
      "authorizationUrl": "https://auth.legaltech.example.com/oauth/authorize",
      "tokenUrl": "https://auth.legaltech.example.com/oauth/token",
      "scopes": ["a2a:tasks:create", "a2a:tasks:read"]
    }
  }
}

Agent Discovery Implementation

# hermes/a2a/discovery.py
import aiohttp
from typing import Dict, Optional

class AgentDiscovery:
    WELL_KNOWN_PATH = "/.well-known/agent.json"
    
    def __init__(self):
        self._cache: Dict[str, Dict] = {}
    
    async def discover(self, agent_url: str) -> Optional[Dict]:
        if agent_url in self._cache:
            return self._cache[agent_url]
        
        card_url = agent_url.rstrip("/") + self.WELL_KNOWN_PATH
        async with aiohttp.ClientSession() as session:
            try:
                async with session.get(card_url, timeout=aiohttp.ClientTimeout(total=10)) as resp:
                    if resp.status == 200:
                        card = await resp.json()
                        self._cache[agent_url] = card
                        return card
            except Exception as e:
                print(f"Could not discover agent at {agent_url}: {e}")
        return None
    
    def can_handle(self, agent_card: Dict, task_description: str) -> float:
        """Score agent suitability for a task (0.0–1.0)"""
        description = task_description.lower()
        score = 0.0
        for skill in agent_card.get("skills", []):
            for tag in skill.get("tags", []):
                if tag.lower() in description:
                    score += 0.3
            for word in description.split():
                if word in skill.get("description", "").lower():
                    score += 0.1
        return min(score, 1.0)

40.4 A2A Protocol API Specification

Core Endpoints

POST   /tasks                    Create a new task
GET    /tasks/{id}               Get task status
PUT    /tasks/{id}/cancel        Cancel a task
GET    /tasks/{id}/artifacts     Get task artifacts
POST   /tasks/{id}/messages      Send a message (multi-turn)
GET    /tasks/{id}/messages      Get message history
GET    /tasks/{id}/stream        SSE stream for task progress
GET    /.well-known/agent.json   Get Agent Card

Create Task Request/Response

// POST /tasks
{
  "message": {
    "role": "user",
    "parts": [
      {"type": "text", "text": "Analyze this contract for compliance risks"},
      {"type": "file", "file": {"name": "contract.pdf", "mimeType": "application/pdf", "bytes": "..."}}
    ]
  },
  "sessionId": "session-12345",
  "metadata": {
    "priority": "high",
    "callback_url": "https://hermes.myapp.com/a2a/callbacks"
  }
}

// 202 Accepted
{
  "id": "task-8f3a92b1-4c2d-4e5f-a123-456789abcdef",
  "status": {
    "state": "submitted",
    "timestamp": "2025-01-15T09:28:00Z"
  },
  "metadata": {"estimated_completion": "2025-01-15T09:33:00Z"}
}

40.5 Hermes A2A Support Status

Hermes 1.7.x A2A Feature Matrix

Configuring A2A in Hermes

# hermes_config.yaml
a2a:
  server:
    enabled: true
    url: "https://hermes.myapp.com"
    skills:
      - id: "research-and-summarize"
        name: "Research and Summarize"
        description: "Deep web research on a topic with a structured summary report"
        tags: ["research", "summarization", "information-retrieval"]
    auth:
      required: true
      schemes: ["bearer"]
  
  client:
    known_agents:
      - name: "legal-agent"
        url: "https://legal-agent.partner.com"
        auth:
          type: "bearer"
          token_env: "LEGAL_AGENT_TOKEN"
      - name: "translation-agent"
        url: "https://translate.service.com"
        auth:
          type: "oauth2"
          client_id_env: "TRANSLATE_CLIENT_ID"
          client_secret_env: "TRANSLATE_CLIENT_SECRET"

40.6 Case Study: Two Hermes Instances Collaborating via A2A

Scenario

User request:
"Analyze this English contract for compliance risks and translate it to Chinese"

Hermes orchestrator breaks down the task:
1. Delegate to Legal Agent (compliance analysis)
2. Delegate to Translation Agent (translation)
3. Merge both results and return to user

A2A Client Implementation

# hermes/a2a/client.py
import asyncio, aiohttp, json
from typing import Any, Dict, AsyncGenerator

class A2AClient:
    def __init__(self, agent_url: str, auth_token: str):
        self.agent_url = agent_url.rstrip("/")
        self.auth_token = auth_token
    
    @property
    def headers(self) -> Dict:
        return {"Authorization": f"Bearer {self.auth_token}", "Content-Type": "application/json"}
    
    async def create_task(self, text: str, files: list = None, session_id: str = None) -> Dict:
        parts = [{"type": "text", "text": text}]
        if files:
            parts.extend({"type": "file", "file": f} for f in files)
        payload = {"message": {"role": "user", "parts": parts}}
        if session_id:
            payload["sessionId"] = session_id
        
        async with aiohttp.ClientSession(headers=self.headers) as session:
            async with session.post(f"{self.agent_url}/tasks", json=payload) as resp:
                resp.raise_for_status()
                return await resp.json()
    
    async def stream_task_progress(self, task_id: str) -> AsyncGenerator[Dict, None]:
        async with aiohttp.ClientSession(headers=self.headers) as session:
            async with session.get(f"{self.agent_url}/tasks/{task_id}/stream") as resp:
                resp.raise_for_status()
                async for line in resp.content:
                    line = line.decode().strip()
                    if line.startswith("data: "):
                        data = json.loads(line[6:])
                        yield data
                        if data.get("type") in ("completed", "failed"):
                            break
    
    async def get_artifacts(self, task_id: str) -> list:
        async with aiohttp.ClientSession(headers=self.headers) as session:
            async with session.get(f"{self.agent_url}/tasks/{task_id}/artifacts") as resp:
                resp.raise_for_status()
                return await resp.json()
    
    async def cancel_task(self, task_id: str):
        async with aiohttp.ClientSession(headers=self.headers) as session:
            async with session.put(f"{self.agent_url}/tasks/{task_id}/cancel") as resp:
                resp.raise_for_status()

Full Orchestration

# orchestrator_demo.py
import asyncio, os
from hermes.a2a.client import A2AClient

async def analyze_contract_with_collaboration(contract_text: str) -> dict:
    legal_client = A2AClient(
        "https://legal-agent.partner.com", os.environ["LEGAL_AGENT_TOKEN"]
    )
    translation_client = A2AClient(
        "https://hermes-translation.myapp.com", os.environ["TRANSLATION_AGENT_TOKEN"]
    )
    
    # Step 1: Launch both tasks in parallel
    print("Launching parallel tasks: legal analysis + translation...")
    legal_task, translation_task = await asyncio.gather(
        legal_client.create_task(
            f"Analyze this contract for compliance risks, focusing on data protection "
            f"and IP clauses:\n\n{contract_text}"
        ),
        translation_client.create_task(
            f"Translate the following contract to Chinese (professional legal translation):\n\n{contract_text}"
        )
    )
    
    # Step 2: Wait for both with streaming progress
    async def stream_until_done(client, task_id, name):
        async for event in client.stream_task_progress(task_id):
            state = event.get("status", {}).get("state", "")
            progress = event.get("status", {}).get("progress", 0)
            print(f"[{name}] {progress*100:.0f}% ({state})")
            if state == "completed":
                return await client.get_artifacts(task_id)
            elif state == "failed":
                raise RuntimeError(f"{name} task failed")
    
    legal_artifacts, translation_artifacts = await asyncio.gather(
        stream_until_done(legal_client, legal_task["id"], "Legal Analysis"),
        stream_until_done(translation_client, translation_task["id"], "Translation")
    )
    
    # Step 3: Extract and merge results
    def get_text(artifacts):
        for artifact in artifacts:
            for part in artifact.get("parts", []):
                if part["type"] == "text":
                    return part["text"]
        return ""
    
    legal_analysis = get_text(legal_artifacts)
    chinese_translation = get_text(translation_artifacts)
    
    import datetime
    final_report = f"""# Contract Analysis Report

## Chinese Translation

{chinese_translation}

---

## Compliance Risk Analysis

{legal_analysis}

---
*Generated by Legal Agent + Translation Agent collaboration*
*Timestamp: {datetime.datetime.now().isoformat()}*
"""
    return {
        "report": final_report,
        "legal_task_id": legal_task["id"],
        "translation_task_id": translation_task["id"]
    }


if __name__ == "__main__":
    sample = """
    SERVICE AGREEMENT
    This Agreement is entered into as of January 1, 2025, by TechCorp Inc.
    and ClientCo Ltd. All client data shall be processed per GDPR...
    """
    result = asyncio.run(analyze_contract_with_collaboration(sample))
    print(result["report"])

40.7 Deep Comparison: A2A vs MCP

Chapter Summary

The A2A protocol fills the multi-Agent collaboration gap and together with MCP forms a complete Agent capability invocation stack:

1. Design background: MCP handles tool invocation; A2A handles Agent delegation—different scopes, complementary roles
2. Core concepts: Task (work unit), Artifact (output), Message (communication)—the three pillars of A2A
3. Agent Card: /.well-known/agent.json is an Agent's "business card," enabling automatic discovery
4. Async by default: A2A natively supports long-running tasks, streaming progress, and Webhook callbacks
5. Hermes support: Version 1.7.x fully supports both A2A Client and Server roles
6. Case study: Two Hermes instances collaborate in parallel to complete a complex legal analysis + translation task

Review Questions

1. Agent Cards are currently static (/.well-known/agent.json). If an Agent's capabilities change dynamically (e.g., certain Skills are disabled under high load), how should Agent Card design accommodate this?
2. In an A2A multi-hop delegation scenario (A→B→C→D), how would you trace the complete call chain? How would you implement distributed tracing?
3. If Agent B needs to ask Agent A clarifying questions mid-task (reversing the delegation direction), is A2A's multi-turn message mechanism sufficient? What are its limitations?
4. Design an "A2A Task Marketplace" where Agents can post tasks needing help from other Agents, and other Agents can bid to accept them. What additional mechanisms would be needed on top of the A2A protocol?