Belajar A2A - Streaming & Push Notifications

#Pendahuluan

Di episode 6 kalian membangun agent nyata dengan a2a-sdk Python dan @a2a-js/sdk TypeScript. Kedua SDK tadi sudah mengantongi fitur streaming dan push notification, tapi kita baru menyentuh permukaannya. Kali ini kita membongkar keduanya sampai ke level wire: bagaimana event task mengalir real-time lewat SSE, bagaimana server memberi tahu client lewat webhook, dan bagaimana sistem tetap bekerja saat koneksi atau webhook bermasalah.

Roadmap episode ini: pertama kita bandingkan dua mode komunikasi — blocking dan streaming. Lalu kita lihat alur SSE di wire, praktik streaming di kedua SDK, konfigurasi push notification, dan terakhir strategi retry serta fallback polling.

#Dua Mode Komunikasi: Blocking vs Streaming

Sejak episode 3 kita mengenal message/send sebagai cara termudah mengirim task. Method ini blocking: client mengirim request, server memproses, lalu mengembalikan task final dalam satu respons HTTP. Nyaman untuk pekerjaan cepat, tetapi bermasalah untuk tugas yang berjalan lama — membayangkan LLM yang berpikir selama dua menit tanpa kabar sama sekali terasa seperti menonton layar loading yang tidak kunjung selesai.

Alternatifnya adalah streaming lewat method message/sendSubscribe. Dengan mode ini server membuka koneksi SSE dan mengalirkan event setiap kali ada perkembangan: status berpindah, pesan progres tiba, artefak parsial terbentuk, hingga task selesai. Perbandingannya:

• Blocking (message/send): satu permintaan, satu respons final. Sederhana, cocok untuk task singkat.
• Streaming (message/sendSubscribe): satu koneksi terbuka, banyak event. Real-time, cocok untuk task panjang.
• Client juga bisa polling (tasks/get) sebagai jalan tengah tanpa koneksi persisten.

#Alur SSE di Wire

SSE (Server-Sent Events) adalah mekanisme HTTP sederhana: server menulis blok data: ke koneksi yang tetap terbuka, dipisahkan baris kosong, dan client membacanya sebagai aliran. Setiap blok di A2A berisi notifikasi JSON-RPC atau respons final. Ini contoh aliran yang diterima client saat task diproses:

data: {"jsonrpc":"2.0","method":"notifications/stream","params":{"kind":"status-update","taskId":"t-102","contextId":"c-7","status":{"state":"working","message":{"role":"agent","parts":[{"kind":"text","text":"Mengumpulkan data dari 3 sumber..."}]}},"final":false}}
data: {"jsonrpc":"2.0","method":"notifications/stream","params":{"kind":"artifact-update","taskId":"t-102","contextId":"c-7","artifact":{"name":"parsial","parts":[{"kind":"text","text":"Sumber 1: 12 referensi"}]},"append":true,"lastChunk":false}}
data: {"jsonrpc":"2.0","id":1,"result":{"kind":"task","id":"t-102","contextId":"c-7","status":{"state":"completed"},"artifacts":[{"name":"laporan","parts":[{"kind":"text","text":"Hasil akhir: 42 poin"}]}]}}

Perhatikan tiga jenis event yang dikenali client:

1. status-update dengan final: false — status antara seperti working, plus pesan progres opsional.
2. artifact-update — potongan artefak yang di-append bertahap, ditandai lastChunk.
3. task dengan id yang sama — task final menutup aliran.

Secara konseptual inilah yang dihasilkan rangkaian update_status dan add_artifact yang kalian tulis di episode 6. Untuk melihatnya langsung tanpa SDK, kalian bisa ikut bergabung ke aliran dengan curl -N:

curl -N -X POST http://localhost:9999/ \
  -H "Content-Type: application/json" \
  -d '{"jsonrpc":"2.0","id":1,"method":"message/sendSubscribe","params":{"message":{"role":"user","parts":[{"kind":"text","text":"Analisis tren"}]}}}'

Dengan -N, curl tidak menunggu aliran berakhir — setiap blok data: langsung dicetak begitu tiba. Berguna untuk debug cepat sebelum menyentuh SDK.

#Streaming di SDK: Python & TypeScript

SDK menyembunyikan semua format SSE di atas. Di Python, A2AClient menyediakan send_message_subscribe yang mengembalikan async iterator; di TypeScript, sendMessageStream mengembalikan async generator:

client.py — streaming dengan send_message_subscribeclient.ts — streaming dengan sendMessageStream

import asyncio
from a2a.client import A2AClient
from a2a.types import Task, TaskStatusUpdateEvent
 
async def main() -> None:
    async with A2AClient(url="http://127.0.0.1:9999") as client:
        async for event in client.send_message_subscribe(
            message={"role": "user", "parts": [{"kind": "text", "text": "Analisis tren"}]}
        ):
            if isinstance(event, TaskStatusUpdateEvent):
                print(f"[status] {event.status.state}")
                if event.status.message:
                    print(event.status.message.parts[0].text)
            elif isinstance(event, Task):
                print(f"[task] {event.id} - {event.status.state}")
 
asyncio.run(main())

import { ClientFactory } from "@a2a-js/sdk/client";
import { v4 as uuidv4 } from "uuid";
 
const client = await new ClientFactory().createFromUrl("http://localhost:4000");
for await (const event of client.sendMessageStream({
  message: { kind: "message", messageId: uuidv4(), role: "user", parts: [{ kind: "text", text: "Analisis tren" }] },
})) {
  if (event.kind === "task") {
    console.log(`[task] ${event.id} - ${event.status.state}`);
  } else if (event.kind === "status-update") {
    console.log(`[status] ${event.status.state}: ${event.status.message?.parts[0].text}`);
  }
}

Pola konsumsi keduanya identik: iterasi event, bedakan tipe, tampilkan. Perbedaan bahasa menjadi tidak relevan — ini kekuatan memakai SDK resmi yang kita tekankan di episode 6.

#Push Notification: Definisi dan Alur

Streaming menuntut koneksi tetap terbuka, yang sulit dipenuhi saat client adalah orchestrator di balik NAT, load balancer, atau scheduler yang hanya "telepon sebentar". Untuk skenario itu A2A menyediakan push notification: client tidak membuka koneksi sama sekali, melainkan memberi tahu server ke mana laporan status harus di-POST.

Konfigurasinya dikirim lewat method tasks/pushNotificationConfig. Parameter url adalah webhook milik client, dan token opsional untuk autentikasi:

{
  "jsonrpc": "2.0",
  "id": 2,
  "method": "tasks/pushNotificationConfig",
  "params": {
    "taskId": "t-102",
    "pushNotificationConfig": {
      "id": "webhook-lead-1",
      "url": "https://orchestrator.example.com/hooks/a2a/t-102",
      "token": "rahasia-hmac"
    }
  }
}

Alur setelahnya berbalik arah. Server mengeksekusi task, lalu setiap kali status berubah ia melakukan HTTP POST ke url di atas dengan payload yang sama persis dengan event SSE (status-update, artifact-update, atau task final). Client cukup menerima POST di endpoint tersebut, memvalidasi token, dan memperbarui status task-nya. Ini cocok untuk arsitektur di mana agent memang harus aktif berkomunikasi dua arah, misalnya orchestrator multi-agent yang memantau puluhan remote agent sekaligus.

#Retry, Policy, dan Fallback Polling

Baik streaming maupun push sama-sama bisa gagal di jaringan. Streaming bisa terputus di tengah jalan; webhook bisa mati atau balas lambat. A2A tidak menyerah pada titik itu — ia menyediakan tiga lapis ketahanan yang saling melengkapi:

1. Retry pada push. Server mengirim ulang POST ke webhook dengan jeda yang membesar (exponential backoff) ketika respons bukan 2xx atau koneksi gagal, hingga batas percobaan. Karena pengiriman bisa berulang, webhook client harus idempotent: task ID dan nomor event dipakai untuk mengabaikan duplikat.
2. Fallback streaming. Saat koneksi SSE terputus, client SDK mencoba rekonneksi dan meminta event yang terlewat. Jika gagal, client bisa melompat ke polling.
3. Polling tasks/get. Paling sederhana dan selalu tersedia. Client menanyakan status task secara berkala:

curl -s -X POST http://localhost:9999/ \
  -H "Content-Type: application/json" \
  -d '{"jsonrpc":"2.0","id":3,"method":"tasks/get","params":{"taskId":"t-102"}}'

Kombinasi ketiganya memberi at-least-once delivery: tidak ada jaminan event tiba tepat sekali, tetapi dijamin tidak ada status yang hilang permanen selama salah satu jalur berhasil. Desain semacam ini wajib untuk agent yang melapor ke orchestrator production.

#Penutup

Episode 7 melengkapi cara agent mengabari perkembangan. Streaming lewat message/sendSubscribe memberi aliran event real-time di atas SSE — dengan tiga tipe event status-update, artifact-update, dan task — sementara push notification lewat tasks/pushNotificationConfig membalik arah komunikasi lewat webhook. Di bawah keduanya ada retry dengan backoff, rekonneksi streaming, dan fallback polling tasks/get yang menjamin tidak ada status yang hilang.

Inti yang harus dibawa pulang:

• message/sendSubscribe membuka koneksi SSE dan mengalirkan event task secara real-time hingga task final.
• Tiga jenis event streaming: status-update (final false), artifact-update (append bertahap), dan task penutup.
• SDK menyembunyikan format SSE; Python dan TypeScript sama-sama memakai pola iterator event.
• Push notification memakai tasks/pushNotificationConfig dengan url webhook dan token opsional, server yang aktif melakukan POST.
• Retry, idempotency, dan fallback polling membentuk at-least-once delivery yang tahan gangguan jaringan.

Streaming dan push membawa konten yang kaya — pesan, artefak, hasil parsial — namun selama ini kita hanya memakai teks polos. Di episode 8 kita akan membedah Content & Structured Output: tipe-tipe Part untuk teks, file, dan data terstruktur berbasis JSON schema, serta cara mengirim hasil terstruktur antar agent tanpa parsing bebas. Sampai jumpa!