← Kembali ke Blog
Oct 22, 2026 · 10 menit baca

Bikin A2A Agent di Python: Agent Lain Bisa Panggil Agent Kamu

A2AAgent ProtocolPythonAI AgentsTutorial

Tim kamu punya agent yang tahu kurs mata uang. Agent tim lain butuh jawaban itu. Pilihannya cuma dua: mereka import module Python kamu dan ikut menanggung dependency-mu, atau kamu bikin HTTP endpoint sendiri plus README yang menjelaskan bentuk request-mu. Dua-duanya jalan sampai agent yang terlibat jadi tiga dan nggak ada yang ingat field-nya namanya apa.

A2A (Agent2Agent) adalah standar yang menggantikan README itu. Satu agent mendeskripsikan dirinya lewat card yang bisa dibaca mesin, agent lain menemukannya, mengirim task, lalu membaca artifact yang balik. Google merilis A2A April 2025 dan menyerahkan stewardship ke Linux Foundation Juni 2025. Spec-nya sampai v1.0 Maret 2026, lalu v1.0.1 (perbaikan bug) Mei 2026.

Bedanya dengan MCP dalam satu baris: MCP menstandarkan cara agent memakai tool, A2A menstandarkan cara satu agent mendelegasikan kerja ke agent lain. Agent yang kamu panggil tetap opaque. Kamu dapat card-nya, state task-nya, dan output-nya, bukan isi perutnya.

Yang perlu disiapkan

  • Python 3.10 atau lebih baru
  • uv (atau pip)
  • curl untuk smoke test

Langkah 1: install SDK

uv add "a2a-sdk[http-server]"

Setara dengan pip install "a2a-sdk[http-server]". SDK ini implement spec 1.0 plus mode kompatibilitas 0.3, dan ada extra untuk gRPC, FastAPI, tracing OpenTelemetry, serta backend SQL (Postgres, MySQL, SQLite).

Langkah 2: tulis logika bisnis dan executor-nya

Mulai dari agent_executor.py. Bagian glue protokolnya punya import path yang harus persis, dan salah sedikit di sini biasanya bikin kehilangan satu sore:

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

RATES = {
    ("USD", "IDR"): 16250.0,
    ("IDR", "USD"): 1 / 16250.0,
    ("EUR", "USD"): 1.09,
}

class CurrencyAgent:
    """Konversi nominal antar pasangan mata uang yang sudah didefinisikan."""

    async def invoke(self, user_request: str) -> str:
        parts = user_request.upper().split()
        if len(parts) != 3:
            return "Kirim request dengan format: <jumlah> <dari> <ke>, contoh 250 USD IDR"
        amount, src, dst = float(parts[0]), parts[1], parts[2]
        rate = RATES.get((src, dst))
        if rate is None:
            return f"Rate untuk {src} ke {dst} belum tersedia"
        return f"{amount:.2f} {src} = {amount * rate:.2f} {dst}"

Logikanya sengaja dibuat membosankan. Nggak ada panggilan LLM, nggak ada API key, jadi kamu bisa lihat protokolnya jalan tanpa harus sambil debug model.

Sekarang executor-nya. Tiga hal yang penting: event pertama yang kamu enqueue harus task-nya, status working ngasih tahu pemanggil kalau kamu sudah mulai, dan artifact itu tempat jawabannya ditaruh.

class CurrencyAgentExecutor(AgentExecutor):
    def __init__(self) -> None:
        self.agent = CurrencyAgent()

    async def execute(self, context: RequestContext, event_queue: EventQueue) -> None:
        if context.current_task:
            task = context.current_task
        else:
            task = new_task_from_user_message(context.message)
            await event_queue.enqueue_event(task)

        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("Processing request..."),
        )

        query = get_message_text(context.message)
        result = (
            await self.agent.invoke(user_request=query)
            if query
            else "No text input was provided."
        )

        await task_updater.add_artifact(
            parts=[new_text_part(text=result, media_type="text/plain")]
        )
        await task_updater.update_status(
            state=TaskState.TASK_STATE_COMPLETED,
            message=new_text_message("Request is completed!"),
        )

    async def cancel(self, context: RequestContext, event_queue: EventQueue) -> None:
        raise NotImplementedError("Cancel is not supported.")

Langkah 3: jalankan servernya

app.py mendefinisikan apa yang bisa dilakukan agent-mu (skill-nya), bagaimana orang lain menemukannya (card-nya), dan di mana dia listen.

import uvicorn
from starlette.applications import Starlette

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 agent_executor import CurrencyAgentExecutor

skill = AgentSkill(
    id="convert_currency",
    name="Convert currency",
    description="Konversi nominal antar dua mata uang pakai tabel rate tetap.",
    input_modes=["text/plain"],
    output_modes=["text/plain"],
    tags=["finance", "currency"],
    examples=["250 USD IDR", "100 EUR USD"],
)

card = AgentCard(
    name="Currency Agent",
    description="Menjawab pertanyaan konversi mata uang untuk agent lain.",
    version="0.1.0",
    default_input_modes=["text/plain"],
    default_output_modes=["text/plain"],
    capabilities=AgentCapabilities(streaming=True, extended_agent_card=False),
    supported_interfaces=[
        AgentInterface(
            protocol_binding="JSONRPC",
            url="http://127.0.0.1:9999",
            protocol_version="1.0",
        )
    ],
    skills=[skill],
)

handler = DefaultRequestHandler(
    agent_executor=CurrencyAgentExecutor(),
    task_store=InMemoryTaskStore(),
    agent_card=card,
)

routes = create_agent_card_routes(card) + create_jsonrpc_routes(handler, "/")
app = Starlette(routes=routes)

if __name__ == "__main__":
    uvicorn.run(app, host="127.0.0.1", port=9999)

url di dalam supported_interfaces harus sama dengan alamat tempat service-nya benar-benar jalan. Kalau bohong, semua client akan membangun request ke alamat yang salah.

Langkah 4: cek card-nya

python app.py

Lalu di terminal kedua:

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

Kamu harus lihat name, entri supportedInterfaces dengan protocolBinding JSONRPC, dan skill-mu. Kalau ini 404, semua langkah setelahnya nggak akan jalan, karena discovery adalah hal pertama yang dilakukan setiap client. Nama field snake_case di Python muncul sebagai camelCase di wire, dan path-nya /.well-known/agent-card.json. Panduan yang ditulis untuk 0.3 masih menyebut agent.json; nama itu dibuang di 0.3.0 dan yang baru didaftarkan di spec saat 1.0.

Langkah 5: panggil seperti agent lain memanggilnya

import asyncio
import httpx

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

BASE_URL = "http://127.0.0.1:9999"

async def main() -> None:
    async with httpx.AsyncClient() as http:
        card = await A2ACardResolver(
            httpx_client=http, base_url=BASE_URL
        ).get_agent_card()
    display_agent_card(card)

    client = await create_client(agent=card, client_config=ClientConfig(streaming=False))
    request = SendMessageRequest(
        message=new_text_message("250 USD IDR", role=Role.ROLE_USER)
    )
    async for chunk in client.send_message(request):
        print(chunk)
    await client.close()

asyncio.run(main())

Iterator non-streaming menghasilkan satu task yang sudah selesai: status COMPLETED, artifact berisi string hasil konversi, dan history percakapannya. Ganti ke ClientConfig(streaming=True) dan panggilan yang sama menghasilkan empat event: task-nya, status update WORKING, artifact update, lalu COMPLETED. Itu bedanya panggilan blocking dengan progres yang bisa kamu tampilkan ke user sambil kerjanya berjalan.

Langkah 6: taruh di internet

A2A agent itu HTTP service biasa, jadi perlakukan seperti itu. Endpoint card tetap publik, karena client harus membacanya dulu sebelum tahu cara autentikasi. Sisanya wajib cek token. Pasang di Starlette dengan cara normal, middleware di sekeliling route JSON-RPC, dan cek otorisasi di dalam tiap pemanggilan executor, bukan dengan asumsi bahwa pemanggil yang boleh baca card otomatis boleh memakainya.

Dua hal yang sering bikin kena di production:

  • InMemoryTaskStore lupa semua task begitu restart. Task yang sedang berstatus INPUT_REQUIRED waktu kamu deploy akan jadi TaskNotFoundError untuk client yang sedang di tengah percakapan. SDK sudah menyediakan SQL task store untuk Postgres, MySQL, dan SQLite; pakai salah satunya.
  • Task ID membocorkan keberadaan data. Kalau kamu balas "not found" untuk task orang lain tapi "permission denied" untuk task sendiri, kamu baru saja mengonfirmasi ID mana yang nyata. Scope semua query task store berdasarkan pemanggil dan balas error not-found yang sama di dua kasus itu.

A2A vs MCP: pilih yang mana

Biasanya butuh dua-duanya, karena pertanyaannya beda.

  • Siapa di ujung sana: MCP ngobrol dengan tool atau resource, A2A ngobrol dengan agent lain.
  • Apa yang terlihat: tool call di MCP transparan, A2A cuma membuka state task dan artifact.
  • Unit kerjanya: MCP punya tool call dan hasilnya, A2A punya task dengan siklus hidup.
  • Transport: MCP pakai stdio dan streamable HTTP, A2A pakai JSON-RPC, gRPC, dan HTTP+JSON.
  • Otorisasi: MCP mengandalkan OAuth 2.1, A2A punya signed agent card lewat ekstensi AP2.

Kalau yang mau kamu panggil itu fungsi yang sebenarnya bisa di-import, itu MCP server. Kalau itu sistem yang punya model sendiri, penilaian sendiri, dan perkiraan sendiri soal berapa lama kerjanya, itu A2A agent.

Kesimpulan

Sekarang kamu bisa berdiriin agent yang ditemukan dan dipanggil agent lain tanpa perlu mereka baca source code-mu. Langkah lanjut: ganti tabel rate-nya dengan panggilan API asli plus timeout, install extra [telemetry] dan lihat apa saja yang ditangkap trace-nya, lalu baca bagian otorisasi di spec sebelum port-nya kamu buka ke luar jaringan.

Referensi

Butuh Bantuan Implementasi?

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

Konsultasi Gratis