Belajar A2A - SDK: Python & TypeScript

pip install a2a-sdk "a2a-sdk[fastapi]" "a2a-sdk[grpc]"

import uvicorn
from starlette.applications import Starlette
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.request_handlers import DefaultRequestHandler
from a2a.server.routes import create_agent_card_routes, create_jsonrpc_routes
from a2a.server.tasks import InMemoryTaskStore, TaskUpdater
from a2a.types import AgentCapabilities, AgentCard, AgentInterface, AgentSkill, TaskState
 
class LeadScorerExecutor(AgentExecutor):
    async def execute(self, context: RequestContext, event_queue: EventQueue) -> None:
        task = new_task_from_user_message(context.message)
        await event_queue.enqueue_event(task)
        updater = TaskUpdater(event_queue=event_queue, task_id=task.id, context_id=task.context_id)
        await updater.update_status(TaskState.TASK_STATE_WORKING, message=new_text_message("Menganalisis lead..."))
        query = get_message_text(context.message)
        result = f"Skor lead untuk '{query or 'kosong'}': 87 (prioritas tinggi)"
        await updater.add_artifact(parts=[new_text_part(text=result, media_type="text/plain")])
        await updater.update_status(TaskState.TASK_STATE_COMPLETED, message=new_text_message("Analisis selesai."))
 
card = AgentCard(
    name="Lead Scorer",
    description="Memberikan skor lead dari deskripsi singkat.",
    url="http://127.0.0.1:9999",
    version="1.0.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=[AgentSkill(id="skor_lead", name="Skor Lead", description="Menghitung skor lead dari deskripsi.", input_modes=["text/plain"], output_modes=["text/plain"])],
)
 
handler = DefaultRequestHandler(agent_executor=LeadScorerExecutor(), task_store=InMemoryTaskStore(), agent_card=card)
routes = create_agent_card_routes(card) + create_jsonrpc_routes(handler, "/")
app = Starlette(routes=routes)
 
uvicorn.run(app, host="127.0.0.1", port=9999)

from a2a.client import A2AClient
 
async def main() -> None:
    async with A2AClient(url="http://127.0.0.1:9999") as client:
        card = await client.get_agent_card()
        print(f"Agent: {card.name}, streaming={card.capabilities.streaming}")
        response = await client.send_message(
            message={"role": "user", "parts": [{"kind": "text", "text": "Lead: PT Nusantara, budget besar"}]}
        )
        task = response.result
        print(f"State: {task.status.state}")
asyncio.run(main())

npm install @a2a-js/sdk express uuid
npm install -D typescript tsx @types/express @types/uuid

import type { AgentExecutor, RequestContext, IExecutionEventBus } from "@a2a-js/sdk";
 
class LeadScorerExecutor implements AgentExecutor {
  async execute(requestContext: RequestContext, eventBus: IExecutionEventBus): Promise<void> {
    const text = requestContext.message.parts.find((p) => p.kind === "text")?.text ?? "";
    const score = `Skor lead untuk '${text}': 87 (prioritas tinggi)`;
    const artifacts = [{ artifactId: "skor-1", parts: [{ kind: "text", text: score }] }];
    await eventBus.publish({
      kind: "task",
      id: requestContext.taskId,
      contextId: requestContext.contextId,
      status: { state: "completed" },
      artifacts,
    });
  }
 
  async cancelTask(taskId: string, eventBus: IExecutionEventBus): Promise<void> {
    console.log(`Membatalkan task ${taskId}`);
  }
}

import express from "express";
import { AGENT_CARD_PATH } from "@a2a-js/sdk";
import { DefaultRequestHandler, InMemoryTaskStore } from "@a2a-js/sdk/server";
import { agentCardHandler, jsonRpcHandler } from "@a2a-js/sdk/server/express";
import { leadScorerCard } from "./card";
import { LeadScorerExecutor } from "./executor";
 
const requestHandler = new DefaultRequestHandler(leadScorerCard, new InMemoryTaskStore(), new LeadScorerExecutor());
const app = express();
app.use(express.json());
app.get(AGENT_CARD_PATH, agentCardHandler(leadScorerCard));
app.post("/", jsonRpcHandler(requestHandler));
app.listen(4000, () => console.log("Lead Scorer siap di :4000"));

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