Learn A2A - SDK: Python & TypeScript
Build real A2A agents with the official SDKs. We dissect the Python a2a-sdk from AgentCard to executor, then the @a2a-js/sdk TypeScript SDK with handler and streaming patterns, complete with client examples.
#Introduction
In episode 5 you mastered the core JSON-RPC methods — message/send, tasks/get, tasks/cancel, tasks/pushNotificationConfig, messages/list — along with their HTTP binding. Now it's time to move up to a more comfortable layer: the official SDKs. With the SDKs, the complexity of serialization, task store, SSE streaming, and Agent Card serving is wrapped up neatly so you can focus on the agent logic. This episode's roadmap: starting with a2a-sdk for Python — install, server structure, to client — then @a2a-js/sdk for TypeScript with handler and streaming patterns.
#Getting to Know the Python a2a-sdk
a2a-sdk is the official Python SDK for A2A, providing both server and client components. Installation is a single line, with optional extras as needed:
pip install a2a-sdk "a2a-sdk[fastapi]" "a2a-sdk[grpc]"The fastapi extra adds web server integration, grpc for the binding we dissect in episode 11, telemetry for OpenTelemetry tracing, and sql for a database-based task store. a2a-sdk also includes all the core protocol types — Task, Message, TextPart, DataPart, FilePart, Artifact — as Pydantic models.
#Python Server: From AgentCard to Executor
As you've learned, an A2A server needs four components: AgentCard, TaskStore, executor, and request handler. Let's build a lead scoring agent:
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)The flow is clear: LeadScorerExecutor is the heart of the logic — the execute method receives a RequestContext and EventQueue, then pushes status through TaskUpdater; every update_status and add_artifact is translated by the SDK into the right JSON-RPC events, including streaming when the client asks for it. For single-function agents, the SDK and ADK also provide a decorator task handler shortcut like @agent.task_method — one function per skill that returns a completed Task. create_agent_card_routes serves the card at /.well-known/agent-card.json, while create_jsonrpc_routes mounts the JSON-RPC methods at the root /.
#Python Client: A2AClient
A2AClient handles Agent Card fetching, JSON-RPC construction, response parsing, and streaming. An example client for the agent above:
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())send_message sends the message/send method and returns the complete task — great for quick jobs. For long-running tasks, there's send_message_subscribe, which opens a stream; we'll cover that specifically in episode 7.
#TypeScript SDK: @a2a-js/sdk
On the JavaScript side, the official SDK is @a2a-js/sdk, running on Node.js with full TypeScript support. The SDK is split into several entry points: the root export contains the shared types (Message, Task, AgentCard), @a2a-js/sdk/server contains the executor and request handler, and @a2a-js/sdk/client contains the client factory.
npm install @a2a-js/sdk express uuid
npm install -D typescript tsx @types/express @types/uuid#TypeScript Server: Handler & Routing
The core of the TypeScript server is the AgentExecutor interface with its execute method accepting a RequestContext and IExecutionEventBus. The TypeScript version of the lead scoring agent:
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}`);
}
}Publishing an event with kind: "task" straight to the completed state is equivalent to the chain of update_status calls in Python. To report progress, add a publish of kind: "status-update" with final: false in the middle of the work. Wiring it into Express is nearly free of "protocol feel":
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"));Note AGENT_CARD_PATH — a constant that points to /.well-known/agent-card.json, the same path as the Python version. This consistency isn't accidental: both SDKs follow the same specification, so Python and TypeScript agents call each other without modification.
#TypeScript Client: Streaming Pattern
@a2a-js/sdk/client provides a ClientFactory that reads the Agent Card from a URL. Streaming is offered through an async generator, so you can use for await naturally:
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;
}This client doesn't care whether the server behind the URL speaks Python or TypeScript — it only knows that the URL serves a valid Agent Card. That's precisely the main point of this episode: the SDKs turn language differences into implementation details, not integration barriers.
#Conclusion
Episode 6 equips you with two official SDKs for building and consuming A2A agents. In Python, a2a-sdk provides the complete component set — AgentCard, DefaultRequestHandler, InMemoryTaskStore, AgentExecutor, TaskUpdater — with a decorator task handler as a shortcut. In TypeScript, @a2a-js/sdk offers the same through AgentExecutor, DefaultRequestHandler, and an async-generator-based streaming client.
Here's the core takeaway:
- Python's
a2a-sdkincludes server and client in one package, with Pydantic models for all protocol types. - The executor is the single place where the agent logic lives; status and artifacts are reported through the event queue and
TaskUpdater. - The decorator task handler trims boilerplate for single-function agents without losing task lifecycle accuracy.
@a2a-js/sdkis split into type, server, and client entry points, with ready-to-use Express middleware.- Both SDKs are fully interoperable because they both implement the A2A specification.
Both SDKs already support streaming and push notifications, but we've only touched the surface. In episode 7 we dissect it in depth: Streaming & Push Notifications — how SSE works for real-time task event streams, pushNotificationConfig configuration, retry strategies, and fallback to polling. See you there!