A2a Protocol

pip install "a2a-sdk[http-server]"   # core + Starlette/SSE server pieces
pip install "a2a-sdk[grpc]"          # gRPC; also: [fastapi] [postgresql] [mysql] [sqlite] [telemetry] [all]
npm install @a2a-js/sdk express      # Node; express is a peer dependency of the server helpers

import uvicorn
from starlette.applications import Starlette
from a2a.helpers import (new_task_from_user_message, new_text_artifact_update_event,
                         new_text_status_update_event)
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
from a2a.types import AgentCapabilities, AgentCard, AgentInterface, AgentSkill, TaskState

class BillingAgent(AgentExecutor):
    async def execute(self, context: RequestContext, event_queue: EventQueue):
        task = context.current_task or new_task_from_user_message(context.message)
        await event_queue.enqueue_event(task)                      # Task first
        await event_queue.enqueue_event(new_text_status_update_event(
            task_id=task.id, context_id=task.context_id,
            state=TaskState.TASK_STATE_WORKING, text="Looking up the invoice..."))
        answer = f"Invoice INV-2041 is paid. You asked: {context.get_user_input()}"
        await event_queue.enqueue_event(new_text_artifact_update_event(
            task_id=task.id, context_id=task.context_id, name="answer", text=answer))
        await event_queue.enqueue_event(new_text_status_update_event(
            task_id=task.id, context_id=task.context_id,
            state=TaskState.TASK_STATE_COMPLETED, text="Done"))

    async def cancel(self, context: RequestContext, event_queue: EventQueue):
        await event_queue.enqueue_event(new_text_status_update_event(
            task_id=context.task_id, context_id=context.context_id,
            state=TaskState.TASK_STATE_CANCELED, text="Canceled"))

card = AgentCard(
    name="Billing Agent", description="Answers questions about invoices.", version="1.0.0",
    supported_interfaces=[AgentInterface(protocol_binding="JSONRPC", url="http://127.0.0.1:9000/")],
    capabilities=AgentCapabilities(streaming=True),
    default_input_modes=["text/plain"], default_output_modes=["text/plain"],
    skills=[AgentSkill(id="invoice-status", name="Invoice status",
                       description="Look up whether an invoice is paid", tags=["billing"],
                       examples=["Is invoice INV-2041 paid?"])],
)
handler = DefaultRequestHandler(agent_executor=BillingAgent(), task_store=InMemoryTaskStore(),
                                agent_card=card)          # agent_card is required in 1.x
app = Starlette(routes=[*create_agent_card_routes(card),
                        *create_jsonrpc_routes(handler, rpc_url="/")])
uvicorn.run(app, host="127.0.0.1", port=9000)

import asyncio
from a2a.client import create_client
from a2a.helpers import new_text_message, get_artifact_text
from a2a.types import Role, SendMessageRequest

async def ask(url: str, text: str) -> str:
    client = await create_client(url)             # fetches the Agent Card, picks a transport
    try:
        answer = ""
        request = SendMessageRequest(message=new_text_message(text, role=Role.ROLE_USER))
        async for chunk in client.send_message(request):   # yields StreamResponse objects
            if chunk.HasField("artifact_update"):
                answer += get_artifact_text(chunk.artifact_update.artifact)
            elif chunk.HasField("message"):
                answer = chunk.message.parts[0].text
        return answer
    finally:
        await client.close()

print(asyncio.run(ask("http://127.0.0.1:9000", "Is invoice INV-2041 paid?")))

# Sequential: research -> write
facts = await ask("http://research-agent.internal:9001", "Summarize 2026 quantum error-correction results")
post = await ask("http://writer-agent.internal:9002", f"Write a 300-word blog post from these notes:\n{facts}")

# Parallel fan-out
market, rivals, feedback = await asyncio.gather(
    ask("http://market-agent.internal:9003", "Analyze market trends"),
    ask("http://competitor-agent.internal:9004", "Analyze competitor products"),
    ask("http://support-agent.internal:9005", "Analyze customer feedback"),
)