← Kembali ke Blog
Aug 19 2026 · 10 menit baca

Bikin A2A Server Pakai Python: Biar AI Agent Bisa Ngobrol Sama Agent Lain

A2AAI AgentsPythonMulti-AgentTutorial

Kebanyakan AI agent hari ini hidup sendiri. Mereka bisa pakai tools, panggil API, dan akses database, tapi nggak bisa ngobrol sama agent lain. Kalau kamu bikin coding agent pakai LangGraph dan orang lain bikin research agent pakai CrewAI, nggak ada cara standar buat mereka kolaborasi.

Protokol Agent2Agent (A2A) jawaban buat masalah ini. Ini standar open yang di-donate oleh Google dan sekarang di bawah Linux Foundation, yang bikin agent dari framework beda bisa komunikasi lewat HTTP. Bayangin USB buat AI agent: satu kabel, banyak device.

Tutorial ini jelasin cara bikin A2A server yang jalan pakai SDK Python resmi, lalu connect ke sana pakai client. Selesai baca, kamu punya dua agent yang saling ngobrol.

Prerequisites

  • Python 3.10 atau lebih baru (cek pakai python3 --version)
  • uv direkomendasikan, atau pip biasa
  • HTTP client buat testing (curl atau browser)
  • Nggak butuh API key LLM buat dasar-dasarnya (sample helloworld cuma pakai logika echo sederhana)

Apa Itu A2A Sebenarnya

A2A mendefinisikan tiga konsep inti:

  1. Agent Card -- dokumen JSON yang mendeskripsikan apa yang bisa dilakukan agent, input/output apa yang diterima, dan di mana menemukannya. Client fetch kartu ini buat discovery.
  2. Task -- satu unit kerja. Client kirim pesan, server buat task, proses, dan kirim hasil. Task bisa sinkron (tunggu hasil) atau asinkron (polling atau push notification).
  3. Transport -- JSON-RPC 2.0 lewat HTTP(S). SDK handle detail protocol supaya kamu fokus ke logika agent.

Protocol ini support streaming via SSE dan push notification buat task yang jalan lama, tapi tutorial ini cover yang dasar: kirim pesan, dapat hasil.

Langkah 1: Install SDK

# Buat direktori project
mkdir a2a-tutorial && cd a2a-tutorial

# Buat virtual environment
python3 -m venv .venv
source .venv/bin/activate

# Install A2A SDK dengan HTTP server support
pip install "a2a-sdk[http-server]"

Extra [http-server] install Starlette dan uvicorn yang dipakai SDK buat serve agent kamu. Per Agustus 2026, versi terbaru adalah 1.1.2.

Langkah 2: Definisikan Agent

Buat file bernama agent.py. Ini tempat logika agent kamu. Buat tutorial ini, agent cuma ngulang pesan user dengan twist.

class EchoAgent:
    """Agent yang ngulang pesan user."""

    async def invoke(self, user_request: str) -> str:
        return f"Echo dari A2A: kamu bilang '{user_request}'"

Di project real, ini tempat kamu wire up panggilan LLM, tool use, atau logika lain. Inti dari A2A adalah protocol-nya nggak peduli apa yang terjadi di dalam invoke. Yang penting format pesan yang masuk dan keluar.

Langkah 3: Bangun Agent Executor

SDK butuh AgentExecutor yang jembatani protocol A2A ke agent kamu. Buat executor.py:

from a2a.helpers import (
    get_message_text,
    new_task_from_user_message,
    new_text_message,
    new_text_part,
)
from a2a.server.agent_execution import AgentExecutor, RequestContext
from a2a.server.events import EventQueue
from a2a.server.tasks import TaskUpdater
from a2a.types import TaskState

from agent import EchoAgent


class EchoAgentExecutor(AgentExecutor):
    def __init__(self):
        self.agent = EchoAgent()

    async def execute(self, context: RequestContext, event_queue: EventQueue):
        # Ambil atau buat task
        if context.current_task:
            task = context.current_task
        else:
            task = new_task_from_user_message(context.message)
            await event_queue.enqueue_event(task)

        # Update status jadi "working"
        task_updater = TaskUpdater(
            event_queue=event_queue,
            task_id=task.id,
            context_id=task.context_id,
        )
        await task_updater.update_status(
            state=TaskState.TASK_STATE_WORKING,
            message=new_text_message("Proses..."),
        )

        # Jalankan agent
        query = get_message_text(context.message)
        result = await self.agent.invoke(user_request=query or "")

        # Kembalikan hasil sebagai artifact
        await task_updater.add_artifact(
            parts=[new_text_part(text=result, media_type="text/plain")]
        )

        # Tandai task selesai
        await task_updater.update_status(
            state=TaskState.TASK_STATE_COMPLETED,
            message=new_text_message("Selesai."),
        )

    async def cancel(self, context: RequestContext, event_queue: EventQueue):
        raise NotImplementedError("Cancel nggak didukung.")

Polanya: terima request, buat atau re-use task, jalankan logika, kirim hasil, tandai selesai. SDK handle semua framing JSON-RPC, manajemen state task, dan transport HTTP.

Langkah 4: Tulis Server

Sekarang wire semuanya di server.py:

import uvicorn

from a2a.server.request_handlers import DefaultRequestHandler
from a2a.server.routes import create_agent_card_routes, create_jsonrpc_routes
from a2a.server.tasks import InMemoryTaskStore
from a2a.types import (
    AgentCapabilities,
    AgentCard,
    AgentInterface,
    AgentSkill,
)
from executor import EchoAgentExecutor
from starlette.applications import Starlette


if __name__ == "__main__":
    # Definisikan apa yang agent ini bisa lakukan
    skill = AgentSkill(
        id="echo",
        name="Echo",
        description="Ngalin pesan user balik.",
        input_modes=["text/plain"],
        output_modes=["text/plain"],
        tags=["echo", "tutorial"],
        examples=["halo", "apa kabar"],
    )

    # Agent Card -- client fetch ini buat discover agent kamu
    agent_card = AgentCard(
        name="Echo Agent",
        description="Agent echo sederhana buat belajar A2A.",
        version="0.1.0",
        default_input_modes=["text/plain"],
        default_output_modes=["text/plain"],
        capabilities=AgentCapabilities(streaming=True),
        supported_interfaces=[
            AgentInterface(
                protocol_binding="JSONRPC",
                url="http://127.0.0.1:9999",
                protocol_version="1.0",
            )
        ],
        skills=[skill],
    )

    # Wire up request handler
    request_handler = DefaultRequestHandler(
        agent_executor=EchoAgentExecutor(),
        task_store=InMemoryTaskStore(),
        agent_card=agent_card,
    )

    # Buat routes
    routes = []
    routes.extend(create_agent_card_routes(agent_card))
    routes.extend(create_jsonrpc_routes(request_handler, "/"))

    app = Starlette(routes=routes)
    uvicorn.run(app, host="127.0.0.1", port=9999)

Yang perlu diperhatikan:

  • AgentCard adalah dokumen discovery. Client hit .well-known/agent.json buat fetch ini.
  • supported_interfaces kasih tahu client cara reach agent ini. Kita pakai JSON-RPC di port 9999.
  • InMemoryTaskStore nyimpen state task di memori. Oke buat development. Di production, ganti pakai PostgreSQL atau MySQL (SDK support keduanya).

Langkah 5: Jalankan Server

python server.py

Kamu harusnya lihat uvicorn mulai:

INFO:     Started server process [12345]
INFO:     Waiting for application startup.
INFO:     Application startup complete.
INFO:     Uvicorn running on http://127.0.0.1:9999

Langkah 6: Test Agent Card

Di terminal terpisah:

curl http://127.0.0.1:9999/.well-known/agent.json | python3 -m json.tool

Ini return Agent Card dalam JSON. Kamu harusnya lihat nama agent, deskripsi, skills, dan info koneksi. Ini cara client yang kompatibel A2A discover apa yang bisa dilakukan agent kamu tanpa baca dokumentasi.

Langkah 7: Kirim Pesan

Test paling simpel adalah panggilan JSON-RPC langsung:

curl -X POST http://127.0.0.1:9999/ \
  -H "Content-Type: application/json" \
  -d '{
    "jsonrpc": "2.0",
    "method": "message/send",
    "params": {
      "message": {
        "role": "user",
        "parts": [
          {"type": "text", "text": "Halo dari curl"}
        ]
      }
    },
    "id": 1
  }'

Response-nya berisi task dengan output echo dari agent. Kamu harusnya lihat sesuatu seperti:

{
  "jsonrpc": "2.0",
  "result": {
    "id": "...",
    "status": {"state": "completed"},
    "artifacts": [
      {
        "parts": [
          {"type": "text", "text": "Echo dari A2A: kamu bilang 'Halo dari curl'"}
        ]
      }
    ]
  }
}

Langkah 8: Bangun Python Client

Buat flow yang lebih realistis, bikin client yang discover agent lalu kirim pesan. Buat client.py:

import asyncio

import httpx
from a2a.client import A2ACardResolver, ClientConfig, create_client
from a2a.helpers import new_text_message
from a2a.types import Role, SendMessageRequest


async def main():
    # 1. Discover agent
    async with httpx.AsyncClient() as http_client:
        resolver = A2ACardResolver(
            httpx_client=http_client,
            base_url="http://127.0.0.1:9999",
        )
        agent_card = await resolver.get_agent_card()
        print(f"Connected to: {agent_card.name}")
        print(f"Skills: {[s.name for s in agent_card.skills]}")

    # 2. Buat client dan kirim pesan
    config = ClientConfig(streaming=False)
    client = await create_client(agent=agent_card, client_config=config)

    message = new_text_message("Halo dari Python client!", role=Role.ROLE_USER)
    request = SendMessageRequest(message=message)

    async for chunk in client.send_message(request):
        print(f"Response: {chunk}")

    await client.close()


asyncio.run(main())

Jalankan:

python client.py

Ini flow A2A yang real: discover lewat Agent Card, lalu komunikasi lewat protocol. Client nggak perlu tahu apa pun soal implementasi server kamu. Yang dibutuhkan cuma Agent Card dan SDK A2A.

Langkah 9: Tambahkan Support Streaming

Kalau kamu nyalain streaming di Agent Card (yang sudah kita lakukan), client bisa terima response saat di-generate. Ganti config client:

config = ClientConfig(streaming=True)
client = await create_client(agent=agent_card, client_config=config)

Streaming berguna kalau agent kamu panggil LLM dan kamu mau tampilin token saat datang, bukan nunggu response full.

Kapan Pakai A2A vs MCP

Ini pertanyaan yang sering muncul, jadi versi singkatnya:

A2A MCP
Tujuan Komunikasi agent-ke-agent Komunikasi agent-ke-tool
Analogi Dua orang ngobrol Orang pakai obeng
Discovery Agent Cards Tool schemas
Use case Agent kamu delegate riset ke agent spesialis Agent kamu panggil API search atau baca file
Dua-duanya dibutuhkan? Ya, mereka saling melengkapi

Dalam praktek, agent mungkin pakai MCP buat akses tools-nya secara internal, dan A2A buat delegate kerja ke agent lain secara external. Mereka bukan standar yang bersaing.

Pertimbangan Production

Beberapa hal yang perlu di-handle sebelum ship:

  1. Ganti InMemoryTaskStore pakai database. SDK support PostgreSQL, MySQL, dan SQLite langsung. In-memory hilang semua saat restart.

  2. Tambahkan authentication. Agent Card support extended card yang cuma terlihat setelah autentikasi. Pakai ini buat gate akses ke skill sensitif.

  3. Jalankan di belakang reverse proxy. Taruh Caddy atau Nginx di depan buat TLS termination, rate limiting, dan logging.

  4. Validasi Agent Card dari pihak ketiga. Setiap agent card dari pihak ketiga adalah input yang nggak terpercaya. Sanitize field kayak name dan description sebelum dipakai di prompt LLM buat cegah prompt injection.

  5. Tambah OpenTelemetry. SDK punya support telemetry opsional. Install pakai pip install "a2a-sdk[telemetry]" dan wire tracing buat debug workflow multi-agent.

Langkah Selanjutnya

  • Jalankan full samples dari a2a-samples repo buat lihat pola yang lebih kompleks
  • Coba A2A Inspector buat inspeksi visual capabilities agent kamu
  • Bangun agent yang pakai MCP (buatan tools) dan A2A (buat delegation ke agent lain) sekaligus
  • Cek kursus DeepLearning.AI tentang A2A buat guided project di berbagai framework

Referensi

Butuh Bantuan Implementasi?

Saya membantu tim mendesain dan membangun infrastruktur cloud scalable, pipeline DevOps, dan sistem production-grade.

Konsultasi Gratis