A server

The three agents in the kit come from one file of 675 lines, written with the Python standard library only. Each agent serves a card, answers both JSON bindings, runs its own logic in a thread, and keeps its tasks in a dictionary. When you finish this section, you can name each part a server must build, and find it in that file and in the reference SDK.

Figure 7.2 follows one SendStreamingMessage through those parts, from the HTTP request to the stream frames and the webhook POST.

The card and one handler per binding

Card route: the GET on /.well-known/agent-card.json that returns the card. The kit's Handler.do_GET answers it from published_card. The SDK's create_agent_card_routes serves the same path, named by AGENT_CARD_WELL_KNOWN_PATH.

Binding handler: one entry point per binding, which parses the request, runs the checks, and calls the same agent methods. The kit has Handler.jsonrpc at /a2a/jsonrpc and Handler.rest under /a2a/rest/, and no gRPC, as gRPC explains. The SDK has create_jsonrpc_routes, create_rest_routes, and GrpcHandler. All three call one DefaultRequestHandler, which is DefaultRequestHandlerV2.

    def jsonrpc(self, raw):
        request_id = None
        if not self.authenticate():
            return None
        try:
            try:
                request = json.loads(raw)
            except ValueError:
                raise A2AError("JSONParseError")
            if not isinstance(request, dict) or request.get("jsonrpc") != "2.0" or "method" not in request:
                raise A2AError("InvalidRequestError")
            request_id = request.get("id")
            self.check_version()
            method = request["method"]
            params = request.get("params") or {}
            agent = self.agent
            if method == "SendMessage":
                return self.write_json(200, {"jsonrpc": "2.0", "id": request_id, "result": agent.send(params)}, "application/json")
            if method == "SendStreamingMessage":
                if not agent.card["capabilities"].get("streaming"):
                    raise A2AError("UnsupportedOperationError")
                stream = queue.Queue()
                agent.send(params, stream)
                self.open_stream()
                return self.pump(stream, lambda event: {"jsonrpc": "2.0", "id": request_id, "result": event})
            …
        except A2AError as error:
            return self.write_json(200, {"jsonrpc": "2.0", "id": request_id, "error": error.jsonrpc()}, "application/json")

Three checks before any work

Authentication runs first. The kit's authenticate answers HTTP 401 with a WWW-Authenticate challenge before any method, on the deployer only. The SDK checks no credential itself: ServerCallContext.user comes from the web framework's middleware. Security schemes says what to check.

The version check runs second. check_version accepts exactly 1.0 and raises VersionNotSupportedError for anything else, because an agent must read an empty value as version 0.3. The SDK's validate_version decorator wraps every JSON-RPC and REST dispatcher method and compares the major version. It does not wrap the gRPC servicer. The version header gives the rules and each SDK's behavior.

The capability check runs third. Before a stream or a push operation, the kit reads card["capabilities"] and raises UnsupportedOperationError or PushNotificationNotSupportedError. The SDK does the same with a validate decorator on on_message_send_stream and on the push methods.

The executor and the task store

Executor: the code that runs the agent for one request and emits its events. The kit's Agent.send validates the message, creates the task in TASK_STATE_SUBMITTED, and starts Agent.work in a thread. work calls behavior, the agent's own logic, and applies each event it yields. The SDK's AgentExecutor.execute receives a RequestContext and an EventQueue, and returns when the work is done or the task enters an interrupted state.

What the executor calls stays hidden from the client. Figure 7.3 draws that line for the three kit agents.

Task store: the record of every task, read by GetTask, ListTasks, and each follow-up message. The kit's Agent.tasks is one dictionary with no owner, and the deployer's bearer token is its only guard. The SDK's TaskStore has save, get, list, and delete, each given the ServerCallContext. InMemoryTaskStore keys tasks by owner, and the default OwnerResolver returns context.user.user_name. Another caller's task is then TaskNotFoundError, but every caller without a user name shares one scope. A server must check authorization on every A2A operation request, so the integrator adds the middleware, as security schemes shows for each SDK.

One event queue for streams and webhooks

The kit's Agent.apply updates the task record. Agent.broadcast then puts a copy of the event on each subscriber queue and calls deliver for each push config. Handler.pump reads one subscriber queue and writes each event as a data: frame. broadcast closes the stream at a terminal state or TASK_STATE_INPUT_REQUIRED, and deliver POSTs the event to the webhook URL with X-A2A-Notification-Token.

The SDK's ActiveTask does the same with two queues. The executor writes to _event_queue_agent. A consumer reads it, saves the task, forwards each event to _event_queue_subscribers, and calls PushNotificationSender.send_notification. Streaming and push notification configs show both paths frame by frame.