A2A quickstart
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.
Prerequisites
Section titled “Prerequisites”- Python 3.10+
- Familiarity with async Python
Step 1: Install the SDK
Section titled “Step 1: Install the SDK”Install the official Python A2A SDK plus uvicorn to serve the app.
pip install a2a-sdk uvicornStep 2: Implement your agent
Section titled “Step 2: Implement your agent”The AgentExecutor holds your business logic, the AgentCard describes your capabilities, and A2AFastAPIApplication provides the HTTP layer.
import uvicornfrom a2a.server.apps.jsonrpc import A2AFastAPIApplicationfrom a2a.server.request_handlers import DefaultRequestHandlerfrom a2a.server.agent_execution import AgentExecutor, RequestContextfrom a2a.server.events import EventQueuefrom a2a.server.tasks import InMemoryTaskStore, TaskUpdaterfrom 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)Step 3: Run it
Section titled “Step 3: Run it”python agent.pyStep 4: Test it
Section titled “Step 4: Test it”Fetch the published Agent Card:
curl http://localhost:8080/.well-known/agent.json | jq .Send a message via JSON-RPC message/send:
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:
npx @a2a-js/inspector http://localhost:8080How it works
Section titled “How it works”The A2A server comprises three key components:
- AgentExecutor — implements business logic, receiving a
RequestContext(the user’s message) and publishing results to anEventQueue. - DefaultRequestHandler — handles protocol details, routing JSON-RPC calls to your executor and managing task lifecycle and storage.
- A2AFastAPIApplication — the HTTP layer that builds a FastAPI/Starlette app with endpoints like
/.well-known/agent.jsonand 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).
Troubleshooting
Section titled “Troubleshooting”| Problem | Fix |
|---|---|
Agent Card not found at /.well-known/agent.json | Verify the correct port and an error-free server startup. |
Tasks remain in working state | Ensure your executor calls complete(), fail(), or cancel() in all code paths, including errors. |
| Authentication fails when Foldspace invokes the agent | Confirm your Agent Card’s authentication.schemes matches what the server validates. |
Next steps
Section titled “Next steps”- Advanced patterns → — LangGraph, TypeScript, Java, native (no-SDK), and streaming.
- Connect to Foldspace → — register your agent in Agent Studio.