How the A2A Protocol Works for Multi-Agent Communication: HTTP-Based Discovery and Task Execution

The A2A protocol enables multi-agent communication through a lightweight HTTP contract where agents discover peers via /.well-known/agent.json, submit tasks via POST /tasks, and poll GET /tasks/{id} to retrieve structured artifacts upon completion.

The A2A (Agent-to-Agent) protocol implemented in the rohitg00/ai-engineering-from-scratch repository provides a minimalist, HTTP-based standard for autonomous agent coordination. This approach eliminates the need for complex RPC frameworks by using standard web verbs to handle peer discovery, work delegation, and asynchronous result retrieval, making it ideal for building horizontally scalable agent swarms.

A2A Protocol Phases and HTTP Interface

The protocol operates through three distinct phases, each mapped to specific HTTP endpoints defined in phases/16-multi-agent-and-swarms/12-a2a-protocol/code/main.py.

Phase 1: Agent Discovery via Well-Known Endpoints

Before delegation, a client agent must discover a peer's capabilities. The A2A protocol mandates hosting a JSON agent card at the well-known URI /.well-known/agent.json.

In main.py (lines 21-31), the AGENT_CARD dictionary defines this self-description schema:

AGENT_CARD = {
    "name": "code-review-agent",
    "version": "1.0.0",
    "protocol_version": "a2a-0.3",
    "skills": ["review-python"],
    "endpoints": {"tasks": "/tasks"},
    "auth": {"type": "none"}  # Extensible to OAuth or mTLS

}

A client performs discovery by issuing GET /.well-known/agent.json, receiving metadata including supported skills, reachable endpoints, and protocol version ("a2a-0.3"). This allows agents to dynamically determine if a peer can handle specific task types before transmitting payloads.

Phase 2: Asynchronous Task Submission

Once a compatible skill is identified, the client submits work via POST /tasks. The handler in main.py (lines 109-115) accepts a JSON payload containing a skill identifier and arbitrary payload data:

# Client submission structure

payload = {
    "skill": "review-python",
    "payload": {"code": "def example():\n    pass\n"}
}

The server generates a unique task ID, initializes the task state as "submitted", and immediately returns this ID to the client. The actual execution occurs asynchronously via TaskStore.create(), which spawns a daemon thread to process the request without blocking the HTTP response.

Phase 3: Result Polling and Artifact Retrieval

Because execution occurs in background threads, clients must poll for completion. The server exposes GET /tasks/{id} (lines 96-106), returning the current task state and any generated artifacts.

A typical polling loop involves repeated requests until the state transitions to "completed" or "failed":

# Polling logic (simplified)

while True:
    task = http_get(f"/tasks/{task_id}")
    if task["state"] in ("completed", "failed"):
        artifact = task["artifact"]  # Typed result data

        break
    time.sleep(0.1)

Core Implementation Details

The reference implementation in main.py demonstrates three critical architectural components that ensure reliable multi-agent communication.

The Agent Card and Protocol Metadata

The agent card serves as the service discovery mechanism, advertising capabilities without requiring a centralized registry. The protocol_version field ("a2a-0.3") ensures forward compatibility, while the auth field (currently "none") reserves schema space for future token-based or mutual-TLS authentication schemes (lines 21-30).

Thread-Safe Task Lifecycle Management

Concurrent task execution requires careful state management. The TaskStore class (lines 38-73) protects mutable task states using threading.Lock, preventing race conditions when multiple agents submit work simultaneously.

The _run method (lines 50-73) handles the lifecycle transitions:

1. Set state to "working"
2. Execute the requested skill (e.g., "review-python")
3. Generate a typed artifact
4. Transition to "completed"

This design supports long-running operations without blocking the main HTTP server thread.

Skill Execution and Typed Artifacts

Results follow a structured schema defined in lines 66-70, wrapping output in a predictable envelope:

task["artifact"] = {
    "type": "structured",  # or "text", "blob", etc.

    "data": {
        "issues": [...],
        "lines": 42
    }
}

In the demo implementation, the "review-python" skill analyzes source code and returns lint-style issues alongside metadata. This typed artifact pattern ensures clients can programmatically consume results regardless of the specific skill executed.

Practical Implementation Walkthrough

Starting the A2A Server

The server runs as a threaded HTTP daemon on port 8765:

from phases/16-multi-agent-and-swarms/12-a2a-protocol/code/main import run_server

server = run_server()  # Non-blocking, runs in daemon thread

Complete Client Interaction

The following implementation demonstrates the full discovery-submission-polling cycle against a running agent:

from urllib.request import Request, urlopen
import json
import time

def http_json(method, url, body=None):
    data = json.dumps(body).encode() if body else None
    req = Request(url, data=data, method=method)
    req.add_header("Content-Type", "application/json")
    with urlopen(req) as resp:
        return json.loads(resp.read().decode())

# 1. Discovery: Fetch agent capabilities

card = http_json("GET", "http://localhost:8765/.well-known/agent.json")
print(f"Discovered {card['name']} with skills: {card['skills']}")

# 2. Task Submission: Delegate code review

task_payload = {
    "skill": "review-python",
    "payload": {"code": "x = 1\nprint(x)\n"}
}
submission = http_json("POST", 
                      f"http://localhost:8765{card['endpoints']['tasks']}", 
                      task_payload)
task_id = submission["task_id"]

# 3. Polling: Retrieve results asynchronously

for _ in range(10):
    task = http_json("GET", f"http://localhost:8765/tasks/{task_id}")
    print(f"State: {task['state']}")
    if task["state"] in ("completed", "failed"):
        print(f"Artifact: {task['artifact']}")
        break
    time.sleep(0.1)

Extending with Custom Skills

Adding capabilities requires modifying the TaskStore._run method (lines 50-73). For example, implementing a text summarization skill:

# Inside TaskStore._run

if task["skill"] == "summarize-text":
    text = task["payload"].get("text", "")
    summary = text.split(".")[0] + "."
    task["artifact"] = {
        "type": "text",
        "data": summary
    }
    task["state"] = "completed"

Clients can then POST {"skill": "summarize-text", "payload": {...}} and receive appropriate artifacts.

Horizontal vs. Vertical Agent Communication

According to the repository's architecture, the A2A protocol facilitates horizontal communication (agent-to-agent delegation), distinguishing it from the MCP (Multi-Channel Protocol) pattern used for vertical agent-to-tool interactions. This separation allows developers to compose systems where high-level agent swarms (A2A) delegate to specialized tool interfaces (MCP) without coupling concerns.

Summary

• Discovery Phase: Agents advertise capabilities via /.well-known/agent.json, enabling dynamic peer detection without centralized registries.
• Asynchronous Execution: The POST /tasks endpoint creates background threads via TaskStore, allowing non-blocking task submission with thread-safe state management using threading.Lock.
• Structured Results: Typed artifacts with consistent schema (type + data) ensure reliable consumption across different skills and agent implementations.
• Protocol Versioning: The "a2a-0.3" version field and extensible auth schema provide upgrade paths for authentication and feature evolution.
• Horizontal Scaling: The HTTP-first design supports stateless, horizontally scalable agent networks unlike tightly-coupled RPC frameworks.

Frequently Asked Questions

What is the A2A protocol?

The A2A (Agent-to-Agent) protocol is an HTTP-based communication standard that allows autonomous software agents to discover each other's capabilities, delegate tasks, and exchange structured results. It uses standard web verbs (GET, POST) and JSON payloads rather than custom RPC frameworks, making it accessible for micro-agent architectures.

How does agent discovery work in A2A?

Agent discovery relies on well-known URIs. An agent hosts a JSON agent card at /.well-known/agent.json containing its name, supported skills, protocol version, and endpoint locations. Potential clients fetch this document to determine compatibility before initiating task delegation, as implemented in lines 21-31 of main.py.

What is the difference between A2A and MCP protocols?

The repository distinguishes between horizontal and vertical communication patterns. A2A handles horizontal agent-to-agent communication (peers delegating work to peers), while MCP (Multi-Channel Protocol) handles vertical agent-to-tool communication (agents invoking specific capabilities or external APIs). Both can coexist in production systems, with A2A forming the inter-agent layer and MCP forming the tool-integration layer.

How are task results returned in the A2A protocol?

Results are returned as typed artifacts through an asynchronous polling mechanism. After submitting a task via POST /tasks, clients poll GET /tasks/{id} until the state becomes "completed". The final response includes an artifact object containing a type field (e.g., "structured", "text") and a data field with the actual results, ensuring type-safe consumption across different agent implementations.

{
  "mcpServers": {
    "instagit": {
      "command": "npx",
      "args": ["-y", "instagit@latest"]
    }
  }
}