A2A Protocol v1.0: Agent-to-Agent Interoperability with Python's a2a-sdk

TL;DR: A2A Protocol v1.0, yang sudah mencapai status stabil pada April 2026 dan di-govern oleh Linux Foundation, adalah standar terbuka untuk komunikasi agent-to-agent. Dengan a2a-sdk 1.2.2, developer Python bisa membangun A2A server dan client dalam hitungan menit. Artikel ini membahas arsitektur, perbandingan dengan MCP, dan kode kerja yang siap dipakai.

---

Masalah yang Dipecahkan A2A

Bayangkan kamu punya dua agent AI yang dibangun di framework berbeda: satu di LangGraph untuk riset pasar, satu lagi di CrewAI untuk menulis draft konten. Tanpa protokol standar, mereka tidak bisa saling bicara. Solusi selama ini adalah kode glue custom — rapuh, tidak reusable, dan harus ditulis ulang setiap kali salah satu agent berubah.

A2A (Agent-to-Agent) Protocol memecahkan masalah ini dengan menyediakan bahasa bersama untuk komunikasi antar-agent. Diperkenalkan Google Cloud pada April 2025 dengan 50+ partner peluncuran (Salesforce, SAP, Atlassian, Cohere, LangChain, dan lainnya), A2A kini sudah mencapai v1.0 dan di-govern oleh Linux Foundation melalui Agent2Agent Project.

Konsepnya analog dengan protokol internet: seperti halnya HTTP yang memungkinkan server dari vendor berbeda saling berkomunikasi, A2A memungkinkan agent dari framework berbeda berkolaborasi tanpa kode integrasi khusus.

A2A vs MCP: Dua Lapisan yang Berbeda

Ini adalah poin paling krusial yang sering disalahpahami: A2A bukan pengganti MCP — keduanya dirancang untuk saling melengkapi.

Dimensi	MCP	A2A
Relasi	Agent → Tool / Data	Agent → Agent
Arah	Vertikal (agent ke sumber daya)	Horizontal (agent ke agent)
Discovery	Negosiasi capability saat koneksi	Agent Card di /.well-known/agent-card.json
State	Stateless (sejak spesifikasi 2026-07-28)	Task lifecycle penuh (8 state)
Transport	stdio, Streamable HTTP	JSON-RPC 2.0, gRPC, HTTP+JSON/REST
Governance	AAIF (Linux Foundation)	Agent2Agent Project (Linux Foundation)

MCP menjawab pertanyaan: “Bagaimana agent memanggil tool atau mengakses data?” Ini adalah protokol vertikal — satu agent berbicara ke database, API, atau filesystem.

A2A menjawab pertanyaan: “Bagaimana agent mendelegasikan tugas ke agent lain yang dibangun di framework berbeda?” Ini adalah protokol horizontal — koordinasi antar-agent tanpa harus tahu detail internal masing-masing.

Pola referensi yang paling umum: agent inventory menggunakan MCP untuk mengecek stok di database-nya sendiri, lalu menggunakan A2A untuk mendelegasikan purchase order ke agent supplier di cloud berbeda. MCP menangani panggilan tool internal. A2A melintasi batas organisasi.

Timeline A2A: Dari Peluncuran ke Stabilitas

Perjalanan A2A tergolong cepat untuk sebuah protokol industri:

• April 2025 — Google mengumumkan A2A dengan 50+ partner enterprise
• Juni 2025 — Didonasikan ke Linux Foundation (open governance)
• Juli 2025 — v0.3: dukungan gRPC, signed agent cards, Python SDK diperluas
• Agustus 2025 — IBM ACP (Agent Communication Protocol) bergabung ke A2A
• September 2025 — AP2 (Agent Payments Protocol) dirilis sebagai ekstensi
• Desember 2025 — Anthropic mendonasikan MCP ke AAIF di bawah Linux Foundation
• April 2026 — A2A v1.0 dirilis: 150+ organisasi pendukung, 6 SDK production-ready
• Mei 2026 — A2A v1.0.1: 24.500+ GitHub stars

Saat ini, enam SDK resmi sudah production-ready: Python, JavaScript/TypeScript, Go, Java, .NET, dan Rust.

Adopsi cloud-native juga sudah matang. Dukungan GA (General Availability) tersedia di:

• Microsoft Copilot Studio dan Azure AI Foundry
• Amazon Bedrock AgentCore Runtime
• Google Vertex AI Agent Builder

Technical Steering Committee A2A mencakup AWS, Cisco, Google, IBM Research, Microsoft, Salesforce, SAP, dan ServiceNow — menjamin governance yang benar-benar vendor-neutral.

Arsitektur a2a-sdk 1.2.2

a2a-sdk adalah Python SDK resmi untuk A2A Protocol, dipublikasikan di PyPI sebagai a2a-sdk. SDK ini mendukung Python 3.10+ dan dirancang async-first.

Instalasi

pip install a2a-sdk fastapi uvicorn

Atau dengan uv (direkomendasikan):

uv add "a2a-sdk[fastapi]"

Ekstra fastapi menyertakan dependensi untuk menjalankan A2A server dengan FastAPI — framework yang sudah familiar bagi developer Django/FastAPI.

Komponen Server

Server A2A terdiri dari empat komponen utama:

1. AgentExecutor — logika bisnis agent (apa yang dilakukan agent saat menerima task)
2. AgentCard — metadata identitas dan kemampuan agent
3. TaskStore — penyimpanan state task (in-memory atau database)
4. Request Handler — penerjemah request JSON-RPC ke AgentExecutor

Komponen Client

Client A2A menggunakan A2AClient atau ClientFactory untuk:

• Menemukan dan membaca Agent Card
• Mengirim task ke server
• Menerima streaming update via SSE
• Mendaftarkan webhook push notification

Kode Kerja: A2A Server dan Client

Mari kita bangun agent yang bisa meringkas teks — contoh sederhana yang menunjukkan seluruh siklus A2A.

1. Agent Card dan AgentExecutor

# server.py
import asyncio
from uuid import uuid4

from a2a.server.agent_execution import AgentExecutor, RequestContext
from a2a.server.events import EventQueue
from a2a.server.request_handlers import DefaultRequestHandler
from a2a.server.tasks import InMemoryTaskStore
from a2a.server.apps import A2AFastAPIApplication
from a2a.types import (
    AgentCard, AgentCapabilities, AgentSkill,
    Task, TaskStatus, TaskState, TaskStatusUpdateEvent,
)
from a2a.utils.message import new_agent_text_message


class SummarizerAgent(AgentExecutor):
    """Agent yang meringkas teks input."""

    async def execute(
        self, context: RequestContext, event_queue: EventQueue
    ) -> None:
        user_text = context.message.parts[0].root.text

        # Laporkan status 'working'
        await event_queue.put(TaskStatusUpdateEvent(
            task_id=context.task_id,
            context_id=context.context_id,
            status=TaskStatus(state=TaskState.WORKING),
            final=False,
        ))

        # Logika bisnis sederhana: ambil 30% kata pertama
        words = user_text.split()
        summary_length = max(len(words) // 3, 5)
        summary = " ".join(words[:summary_length]) + "..."

        # Kirim hasil akhir
        await event_queue.put(Task(
            id=context.task_id,
            context_id=context.context_id,
            status=TaskStatus(
                state=TaskState.COMPLETED,
                message=new_agent_text_message(
                    f"Summary: {summary}",
                    context.context_id,
                    context.task_id,
                ),
            ),
        ))

    async def cancel(
        self, context: RequestContext, event_queue: EventQueue
    ) -> None:
        await event_queue.put(Task(
            id=context.task_id,
            context_id=context.context_id,
            status=TaskStatus(state=TaskState.CANCELED),
        ))


# Definisikan Agent Card — identitas agent yang dipublikasikan
agent_card = AgentCard(
    name="Summarizer Agent",
    description="Meringkas teks input menjadi versi yang lebih pendek.",
    url="http://localhost:8000",
    version="1.0.0",
    capabilities=AgentCapabilities(
        streaming=True,
        pushNotifications=False,
    ),
    skills=[
        AgentSkill(
            id="summarize",
            name="Summarize Text",
            description="Mengambil teks panjang dan mengembalikan ringkasan pendek.",
            tags=["text", "summarization"],
        ),
    ],
    default_input_modes=["text/plain"],
    default_output_modes=["text/plain"],
)

# Rangkai komponen
handler = DefaultRequestHandler(
    agent_executor=SummarizerAgent(),
    task_store=InMemoryTaskStore(),
)

app = A2AFastAPIApplication(agent_card, handler).build()

2. Menjalankan Server

uv run uvicorn server:app --host 0.0.0.0 --port 8000

Server akan:

• Mengekspos Agent Card di http://localhost:8000/.well-known/agent-card.json
• Menerima task via JSON-RPC 2.0 di endpoint /a2a
• Mengelola task lifecycle (submitted → working → completed)

3. Client untuk Mengirim Task

# client.py
import asyncio
from a2a.client import A2AClient


async def main():
    async with A2AClient(url="http://localhost:8000") as client:
        # Baca Agent Card
        card = await client.get_agent_card()
        print(f"Agent: {card.name}")
        print(f"Skills: {[s.name for s in card.skills]}")

        # Kirim task
        response = await client.send_message(
            message={
                "role": "ROLE_USER",
                "parts": [
                    {
                        "kind": "text",
                        "text": (
                            "A2A Protocol adalah standar terbuka untuk komunikasi "
                            "antar-agent yang memungkinkan agent dari berbagai "
                            "framework saling berkolaborasi tanpa integrasi khusus."
                        ),
                    }
                ],
            }
        )

        task = response.result
        if task.status.state == "COMPLETED":
            print(f"Result: {task.status.message.parts[0].root.text}")


asyncio.run(main())

4. Streaming dengan ClientFactory

Untuk agent yang memproses task lama, gunakan pola streaming:

# streaming_client.py
import asyncio
from a2a.client import ClientFactory, create_text_message_object
from a2a.types import TaskState


async def stream_response():
    client = await ClientFactory.connect("http://localhost:8000")
    try:
        message = create_text_message_object(
            content="Jelaskan A2A Protocol secara detail."
        )

        async for event in client.send_message(message):
            if isinstance(event, tuple):
                task, update = event
                if task.status.state == TaskState.WORKING:
                    print(f"[processing] {task.status.message.parts[0].root.text}")
                elif task.status.state == TaskState.COMPLETED:
                    print(f"[done] {task.status.message.parts[0].root.text}")
                    break
    finally:
        await client.close()


asyncio.run(stream_response())

Task Lifecycle

Setiap task dalam A2A bergerak melalui state machine yang terdefinisi:

SUBMITTED ──► WORKING ──► COMPLETED (terminal)
                 │
                 ├──► INPUT_REQUIRED ──► WORKING ──► COMPLETED
                 ├──► AUTH_REQUIRED
                 ├──► FAILED (terminal)
                 ├──► CANCELED (terminal)
                 └──► REJECTED (terminal, agent menolak)

Di a2a-sdk, state direpresentasikan sebagai enum TaskState dengan nilai SCREAMING_SNAKE_CASE (misal TaskState.WORKING, TaskState.COMPLETED). AgentExecutor melaporkan transisi state dengan menempatkan TaskStatusUpdateEvent (untuk update intermediate) atau Task (untuk state terminal) ke event_queue.

A2A dan Django: Skenario Produksi

Developer Django mungkin bertanya: “Bagaimana A2A terintegrasi dengan proyek Django saya?” Berikut pola umum:

A2A Server sebagai Service Terpisah

Pendekatan paling bersih: jalankan A2A server sebagai service Python terpisah yang mengimpor logika bisnis dari Django app.

┌─────────────────┐      JSON-RPC 2.0       ┌──────────────────┐
│  A2A Client      │ ──────────────────────► │  A2A Server       │
│  (agent lain)    │                         │  (FastAPI + a2a)  │
└─────────────────┘                         └────────┬─────────┘
                                                     │
                                           ┌─────────▼─────────┐
                                           │  Django Service    │
                                           │  Layer (import)    │
                                           └───────────────────┘

AgentExecutor cukup memanggil service layer Django. Karena A2A server berbasis FastAPI, ia bisa berbagi database dan koneksi Redis yang sama dengan Django.

A2A Client dari Django Task Queue

Di sisi konsumen, kamu bisa memanggil A2A agent remote dari Celery task atau background job Django:

# tasks.py — celery task yang memanggil A2A agent remote
import asyncio
from celery import shared_task
from a2a.client import A2AClient


@shared_task
def call_research_agent(query: str) -> str:
    """Memanggil A2A research agent dari Celery task."""

    async def _call():
        async with A2AClient(
            url="http://research-agent.internal:8000"
        ) as client:
            response = await client.send_message(
                message={
                    "role": "ROLE_USER",
                    "parts": [{"kind": "text", "text": query}],
                }
            )
            task = response.result
            if task.status.state == "COMPLETED":
                return task.status.message.parts[0].root.text
            raise RuntimeError(f"Task failed: {task.status.state}")

    return asyncio.run(_call())

Keamanan dan Agent Card yang Ditandatangani

A2A v1.0 memperkenalkan signed agent cards — Agent Card yang ditandatangani secara kriptografis menggunakan ProtoJSON canonicalization. Agent penerima bisa memverifikasi bahwa card benar-benar diterbitkan oleh pemilik domain, bukan hasil man-in-the-middle.

Ini penting secara operasional ketika kamu mendelegasikan tugas ke agent yang tidak kamu bangun sendiri. Di lingkungan enterprise, signed agent cards dikombinasikan dengan mTLS atau OAuth 2.0 untuk otentikasi dua arah.

Kesimpulan

A2A Protocol v1.0 telah matang menjadi standar industri yang layak dipakai di produksi. Dengan governance vendor-neutral di bawah Linux Foundation, dukungan dari tiga cloud utama (AWS, Azure, GCP), dan SDK Python yang stabil, tidak ada alasan untuk terus menulis kode glue custom untuk komunikasi antar-agent.

Takeaway utama:

1. A2A dan MCP itu komplementer — bukan kompetitor. Satu untuk agent-to-tool, satu untuk agent-to-agent.
2. a2a-sdk 1.2.2 siap produksi — async-first, FastAPI-native, dengan dukungan JSON-RPC, gRPC, dan REST.
3. Task lifecycle yang eksplisit — agent selalu tahu state task yang didelegasikan, dari submitted hingga completed atau failed.
4. Adopsi enterprise sudah luas — 150+ organisasi, SDK di 6 bahasa, dukungan GA di tiga cloud utama.

Sumber Daya Lebih Lanjut

• Dokumentasi resmi A2A Protocol
• Spesifikasi v1.0
• GitHub a2a-python SDK
• PyPI a2a-sdk
• A2A dan MCP — penjelasan resmi
• DeepLearning.AI Course: A2A Protocol
• Agent2Agent Project (Linux Foundation)