Skip to content

A2A quickstart

Open in ChatGPT Open in Claude

This quickstart builds a Travel Booking A2A server agent in Python with the official A2A SDK, then runs and tests it locally. For the concepts behind A2A, start with A2A server agents.

  • Python 3.10+
  • Familiarity with async Python

Install the official Python A2A SDK plus uvicorn to serve the app.

Terminal window
pip install a2a-sdk uvicorn

The AgentExecutor holds your business logic, the AgentCard describes your capabilities, and A2AFastAPIApplication provides the HTTP layer.

agent.py
import uvicorn
from a2a.server.apps.jsonrpc import A2AFastAPIApplication
from a2a.server.request_handlers import DefaultRequestHandler
from a2a.server.agent_execution import AgentExecutor, RequestContext
from a2a.server.events import EventQueue
from a2a.server.tasks import InMemoryTaskStore, TaskUpdater
from a2a.types import (
AgentCard,
AgentCapabilities,
AgentSkill,
Part,
TextPart,
)
class TravelBookingExecutor(AgentExecutor):
"""Handles flight and hotel search requests."""
async def execute(self, context: RequestContext, event_queue: EventQueue) -> None:
updater = TaskUpdater(event_queue, context.task_id, context.context_id)
await updater.start_work()
user_input = context.get_user_input()
result = await self._search_travel_options(user_input)
await updater.add_artifact(parts=[Part(root=TextPart(text=result))])
await updater.complete()
async def cancel(self, context: RequestContext, event_queue: EventQueue) -> None:
updater = TaskUpdater(event_queue, context.task_id, context.context_id)
await updater.cancel()
async def _search_travel_options(self, query: str) -> str:
# Replace with your actual logic — call an LLM, query APIs, etc.
return f"Found 3 flight options and 5 hotels matching: {query}"
# --- Agent Card: describes what your agent can do ---
agent_card = AgentCard(
name="Travel Booking Agent",
description="Searches flights, hotels, and packages based on travel preferences",
url="http://localhost:8080/",
version="1.0.0",
capabilities=AgentCapabilities(streaming=True),
default_input_modes=["text/plain"],
default_output_modes=["text/plain", "application/json"],
skills=[
AgentSkill(
id="search-flights",
name="Search Flights",
description="Find available flights between destinations",
tags=["travel", "flights", "booking"],
examples=["Find flights from NYC to London next Friday"],
),
AgentSkill(
id="search-hotels",
name="Search Hotels",
description="Find hotels at a destination with filters",
tags=["travel", "hotels", "accommodation"],
examples=["Find 4-star hotels in Paris for 3 nights"],
),
],
)
# --- Wire up and start the server ---
request_handler = DefaultRequestHandler(
agent_executor=TravelBookingExecutor(),
task_store=InMemoryTaskStore(),
)
app = A2AFastAPIApplication(
agent_card=agent_card,
http_handler=request_handler,
).build()
if __name__ == "__main__":
uvicorn.run(app, host="0.0.0.0", port=8080)
Terminal window
python agent.py

Fetch the published Agent Card:

Terminal window
curl http://localhost:8080/.well-known/agent.json | jq .

Send a message via JSON-RPC message/send:

Terminal window
curl -X POST http://localhost:8080/ \
-H "Content-Type: application/json" \
-d '{
"jsonrpc": "2.0",
"id": "test-001",
"method": "message/send",
"params": {
"message": {
"role": "user",
"parts": [{"kind": "text", "text": "Find flights from SF to Tokyo next week"}],
"messageId": "msg-001"
}
}
}'

Validate protocol compliance with the A2A Inspector:

Terminal window
npx @a2a-js/inspector http://localhost:8080

The A2A server comprises three key components:

  1. AgentExecutor — implements business logic, receiving a RequestContext (the user’s message) and publishing results to an EventQueue.
  2. DefaultRequestHandler — handles protocol details, routing JSON-RPC calls to your executor and managing task lifecycle and storage.
  3. A2AFastAPIApplication — the HTTP layer that builds a FastAPI/Starlette app with endpoints like /.well-known/agent.json and the JSON-RPC POST.

Request flow: Client → A2AFastAPIApplication (HTTP) → DefaultRequestHandler (task lifecycle) → your executor → TaskUpdater publishes status/artifacts to the EventQueue → events return as an SSE stream (streaming) or a single JSON-RPC response (non-streaming).

ProblemFix
Agent Card not found at /.well-known/agent.jsonVerify the correct port and an error-free server startup.
Tasks remain in working stateEnsure your executor calls complete(), fail(), or cancel() in all code paths, including errors.
Authentication fails when Foldspace invokes the agentConfirm your Agent Card’s authentication.schemes matches what the server validates.