Building an A2A Server With the Python a2a-sdk: AgentExecutor and EventQueue

Agent and AI engineering · Practical technical guide · 2 November 2026

An order-status A2A server built on a2a-sdk 1.3.0: AgentExecutor.execute and cancel, RequestContext, EventQueue and TaskUpdater, DefaultRequestHandler and the route factories, run and probed, plus measured edge cases for missing final events, late artifacts and cancellation.

Where is order BBS-1042

Jenya asks the Belgavi's Book Store help desk where her order is. The help-desk agent does not track parcels itself; it asks an order-status agent over A2A, which looks up the courier steps and streams them back. This article builds that order-status agent with the Python a2a-sdk, then breaks it on purpose: an executor that forgets its final event, one that adds an artifact after completing, and a cancel that arrives in the middle of the lookup.

Everything here is pinned to a2a-sdk 1.3.0 (PyPI, released 27 October 2026; installed and run with Python 3.13 on 2 November 2026). The SDK implements A2A protocol 1.0 with a 0.3 compatibility mode; its 1.0 line changed the server API, so 0.3-era tutorial code does not run unchanged.

The pieces in a2a-sdk 1.3.0

AgentExecutor is the class you implement: async def execute(self, context: RequestContext, event_queue: EventQueue) -> None and async def cancel(self, context, event_queue) -> None.
RequestContext carries the incoming message, task_id, context_id, current_task (set when a message continues an existing task), the call context, metadata and requested extensions.
EventQueue receives what the executor produces: either one Message, or a Task followed by TaskStatusUpdateEvent and TaskArtifactUpdateEvent objects.
TaskUpdater wraps the queue with start_work, add_artifact, requires_input, complete, failed, cancel and friends. It raises if you update the status of a task that already reached a terminal state.
DefaultRequestHandler (in 1.3.0 an alias for DefaultRequestHandlerV2) runs the executor, consumes the queue, persists to a TaskStore such as InMemoryTaskStore, and fans events out to clients. Since 1.0 it takes the agent_card in its constructor.
Routes: the 1.0 migration guide says the A2AStarletteApplication and FastAPI wrapper classes were removed. You compose create_agent_card_routes, create_jsonrpc_routes or create_rest_routes into your own Starlette or FastAPI app.
Types are Protobuf classes now, and enum values are SCREAMING_SNAKE_CASE, such as TaskState.TASK_STATE_COMPLETED.

The order-status server

Run with a2a-sdk[http-server] 1.3.0, uvicorn and Python 3.13 on 2 November 2026.

import asyncio

import uvicorn
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
from starlette.applications import Starlette

ORDERS = {'BBS-1042': ['packed', 'handed to courier', 'out for delivery']}


class OrderStatusExecutor(AgentExecutor):
    async def execute(self, context: RequestContext, event_queue: EventQueue) -> None:
        task = context.current_task
        if task is None:
            task = new_task_from_user_message(context.message)
            await event_queue.enqueue_event(task)          # client sees SUBMITTED
        updater = TaskUpdater(event_queue, task.id, task.context_id)
        try:
            await updater.start_work()                     # WORKING
            order_id = get_message_text(context.message).strip().upper()
            steps = ORDERS.get(order_id)
            if steps is None:
                await updater.requires_input(new_text_message(
                    'Which order? Send an id such as BBS-1042.',
                    context_id=task.context_id, task_id=task.id))
                return                                     # interrupted, not terminal
            for i, step in enumerate(steps):
                await asyncio.sleep(0.2)                   # an await point: cancel lands here
                await updater.add_artifact(
                    [new_text_part(step + '; ')], artifact_id='status', name='order-status',
                    append=i > 0, last_chunk=i == len(steps) - 1)
            await updater.complete()                       # terminal: always the last event
        except Exception as exc:
            await updater.failed(new_text_message(
                'Lookup failed: ' + type(exc).__name__, context_id=task.context_id, task_id=task.id))

    async def cancel(self, context: RequestContext, event_queue: EventQueue) -> None:
        await TaskUpdater(event_queue, context.task_id, context.context_id).cancel()


card = AgentCard(
    name='Order status agent',
    description="Belgavi's Book Store: where is my order?",
    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='order_status', name='Order status', description='Tracks a book order by id',
                       tags=['orders'], examples=['BBS-1042'])],
)
handler = DefaultRequestHandler(agent_executor=OrderStatusExecutor(), task_store=InMemoryTaskStore(),
                                agent_card=card)
app = Starlette(routes=[*create_agent_card_routes(card), *create_jsonrpc_routes(handler, '/')])

if __name__ == '__main__':
    uvicorn.run(app, host='127.0.0.1', port=9999)

The executor follows the pattern of the official hello-world sample: enqueue the Task first, move it to WORKING, stream artifact chunks with append and last_chunk, and end with exactly one terminal or interrupted status. The except Exception turns lookup errors into a FAILED status; it deliberately does not catch asyncio.CancelledError, which is how cancellation arrives.

What the client sees

With the server running, a GET of /.well-known/agent-card.json returned the card, and a JSON-RPC SendMessage with the text BBS-1042 returned the finished task. Captured on 2 November 2026, trimmed:

Captured response, trimmed (ids and timestamps removed).

{"result": {"task": {
  "status": {"state": "TASK_STATE_COMPLETED"},
  "artifacts": [{"artifactId": "status", "name": "order-status", "parts": [
    {"text": "packed; "}, {"text": "handed to courier; "}, {"text": "out for delivery; "}]}],
  "history": [{"messageId": "m1", "role": "ROLE_USER", "parts": [{"text": "BBS-1042"}]}]}},
 "id": 1, "jsonrpc": "2.0"}

The same request as SendStreamingMessage produced six server-sent events: the Task in SUBMITTED, a WORKING status update, three artifact updates and a COMPLETED status update. A message with the text "no idea" returned a task in TASK_STATE_INPUT_REQUIRED carrying the agent question. The blocking call waits for a terminal or interrupted state, while the stream shows every step.

Rules the executor has to keep

Enqueue the Task before any status or artifact update, or send exactly one Message and nothing else. The 1.0 handler raises InvalidAgentResponseError for an update before the Task, a Message in task mode, or a second Message.
End every execute() with a terminal status (COMPLETED, FAILED, CANCELED, REJECTED) or an interrupted one (INPUT_REQUIRED, AUTH_REQUIRED), then return. The docstring says the executor SHOULD do this, and the handler does not enforce it.
Do not touch the context or the queue after execute() returns.
Let unhandled exceptions surface: the handler records FAILED. Catch only what you can turn into a meaningful failure message.
Make cancel() write CANCELED, and make the work itself stoppable.

Edge cases measured in 1.3.0

The same executor shape was then driven in-process through DefaultRequestHandler with each mistake. Results:

Executor behaviour	Streaming client	Blocking client	Stored task
Returns without a final status	stream closes after the last artifact	returns the task in WORKING	WORKING until something cancels it
Never returns	stream stays open until the client times out	blocks	WORKING
Artifact added after complete()	the late artifact is delivered after COMPLETED	result lists one artifact	both artifacts saved
Raises an exception	request ends with the error	request ends with the error	FAILED
cancel() raises while execute() runs	FAILED; the CancelTask caller gets the error	FAILED; the CancelTask caller gets the error	FAILED
cancel() raises after execute() returned	already closed	already returned	unchanged; the caller gets the error

The late artifact is the surprising one: TaskUpdater guards status updates after a terminal state but not add_artifact, and in this run nothing downstream dropped it, whether it was enqueued immediately after complete() or 0.3 seconds later. Treat it as observed behaviour of this version rather than a contract, and do not rely on either outcome. Guard it in your own code.

Cancellation is cooperative

On CancelTask the handler awaits your cancel(), so it can still write CANCELED to the open queue, and then cancels the asyncio task that runs execute(). Code that is suspended at an await receives CancelledError and stops. In the measured run, a lookup that awaited between steps stopped after 1 of 6 steps. The same lookup as one blocking function inside asyncio.to_thread also reported CANCELED, but its worker thread ran all 6 steps anyway. Python cannot interrupt a thread, so blocking work needs its own cancel flag, checked between steps and set from cancel().

The official hello-world sample raises NotImplementedError from cancel(). In the measured runs, a cancel while execute() was running returned an error to the CancelTask caller and left the task FAILED, not CANCELED; for a task whose execute() had already returned, the caller got the error and the task kept its WORKING or INPUT_REQUIRED state.

Task store and restarts

InMemoryTaskStore lives in the process: a restart forgets every task, and a client that polls with GetTask afterwards finds nothing. The SDK ships a SQLAlchemy-based DatabaseTaskStore behind the sql, postgresql, mysql and sqlite extras, which keeps task state across restarts. Persisting state is not the same as resuming work: a task that was WORKING when the process died stays WORKING in the database unless your own recovery code re-runs or fails it. The restart behaviour was not measured here; the lab treats it as a stated model assumption.

What the companion lab shows

The lab models the event sequence for the order-status task and what each client sees. At the defaults the stream carries 6 events and ends COMPLETED. Forgetting the final status leaves the task WORKING with no hang; never returning leaves the stream open until the client gives up. Adding an artifact after completion delivers 7 events, one of them after the terminal state, and an EventQueue wrapper of your own that drops such events brings it back to 6. A cancel after the first chunk ends CANCELED with 4 events. With the lookup in asyncio.to_thread it leaves 2 orphaned steps, and with a cancel flag it leaves 1. A restart loses the task with the in-memory store and strands it in WORKING with a database store.

How to verify your executor

Drive DefaultRequestHandler in-process in tests and assert the full event sequence, not just the final state.
Assert that every code path ends with exactly one terminal or interrupted status.
Cancel mid-run and assert that no work continues afterwards, for example with a counter the work increments.
Put a timeout on every streaming call, so a stuck executor fails a test instead of hanging it.

Sources and scope

Checked against these sources on 2 November 2026. a2a-sdk[http-server] 1.3.0 was installed into a throwaway venv with Python 3.13 on 2 November 2026. The server shown was run under uvicorn and probed with curl: Agent Card, SendMessage, SendStreamingMessage and an input-required request. The responses quoted are captured output, trimmed. The edge-case table comes from a separate in-process harness that drove DefaultRequestHandler with deliberately faulty executors. Restart behaviour was not measured. The lab is a deterministic model of these sequences.

In a2a-sdk 1.3.0 an A2A server is an AgentExecutor plus DefaultRequestHandler, a task store and route factories in your own Starlette or FastAPI app. Enqueue the Task first, end every path with one terminal or interrupted status, guard against emitting after the end yourself, and make cancellation reach blocking work, because the handler only cancels at await points.