Package org.a2aproject.sdk.server.requesthandlers

Class DefaultRequestHandler

java.lang.Object
org.a2aproject.sdk.server.requesthandlers.DefaultRequestHandler
All Implemented Interfaces:
RequestHandler
@ApplicationScoped public class DefaultRequestHandler extends Object implements RequestHandler
Central request orchestrator that coordinates transport requests with agent execution, task persistence, event routing, and push notifications.

This class is the core of the A2A server runtime. It receives requests from transport layers (JSON-RPC, gRPC, REST), executes user-provided AgentExecutor logic asynchronously, manages event queues for response streaming, and ensures task state is persisted through TaskStore.

Architecture Overview

 Transport Layer (JSON-RPC/gRPC/REST)
     ↓ calls DefaultRequestHandler methods
 DefaultRequestHandler (orchestrates)
     ↓
 ┌─────────────┬──────────────┬─────────────────┬──────────────────┐
 │ AgentExecutor│  TaskStore   │  QueueManager   │ PushNotification │
 │ (user logic) │ (persistence)│ (event routing) │ (notifications)  │
 └─────────────┴──────────────┴─────────────────┴──────────────────┘

Request Flow - Blocking Mode (onMessageSend)

  1. Transport calls onMessageSend(MessageSendParams, ServerCallContext)
  2. Initialize TaskManager and RequestContext
  3. Create or tap EventQueue via QueueManager
  4. Execute AgentExecutor.execute(RequestContext, AgentEmitter) asynchronously in background thread pool
  5. Consume events from queue on Vert.x worker thread via EventConsumer
  6. For blocking=true: wait for agent completion and full event consumption
  7. Return Task or Message to transport
  8. Cleanup queue and agent future in background

Request Flow - Streaming Mode (onMessageSendStream)

  1. Transport calls onMessageSendStream(MessageSendParams, ServerCallContext)
  2. Initialize components (same as blocking)
  3. Execute AgentExecutor.execute(RequestContext, AgentEmitter) asynchronously
  4. Return Flow.Publisher<StreamingEventKind> immediately
  5. Events stream to client as they arrive in the queue
  6. On client disconnect: continue consumption in background (fire-and-forget)
  7. Cleanup after streaming completes

Queue Lifecycle Management

  • QueueManager.createOrTap(String) creates a MainQueue (new task) or ChildQueue (resubscription)
  • Agent enqueues events on background thread via EventQueue.enqueueEvent(Event)
  • EventConsumer polls and processes events on Vert.x worker thread
  • Queue closes automatically on final event (COMPLETED/FAILED/CANCELED)
  • Cleanup waits for both agent execution AND event consumption to complete

Threading Model

Important: Avoid blocking operations on Vert.x worker threads - they are limited and shared across all requests.

Blocking vs Streaming

  • Blocking (configuration.blocking=true): Client waits for first event or final task state
  • Streaming: Client receives events as they arrive via reactive streams
  • Both modes support fire-and-forget (agent continues after client disconnect)
  • Configurable timeouts via a2a.blocking.agent.timeout.seconds, a2a.blocking.consumption.timeout.seconds, and a2a.blocking.reconciliation.timeout.seconds

CDI Dependencies

This class is @ApplicationScoped and automatically injects:

Extension Strategy

Users typically don't replace DefaultRequestHandler. Instead, provide custom implementations of its dependencies via CDI:
  • AgentExecutor (required) - Your agent business logic
  • TaskStore (@Alternative @Priority) - Database persistence (see extras/task-store-database-jpa)
  • QueueManager (@Alternative @Priority) - Replication support (see extras/queue-manager-replicated)
  • PushNotificationSender (@Alternative @Priority) - Custom notification delivery
See Also:

  • Constructor Details

    • DefaultRequestHandler

      protected DefaultRequestHandler()
      No-args constructor for CDI proxy creation. CDI requires a non-private constructor to create proxies for @ApplicationScoped beans. All fields are initialized by the @Inject constructor during actual bean creation.

    • DefaultRequestHandler

      @Inject public DefaultRequestHandler(AgentExecutor agentExecutor, TaskStore taskStore, QueueManager queueManager, PushNotificationConfigStore pushConfigStore, MainEventBusProcessor mainEventBusProcessor, Executor executor, Executor eventConsumerExecutor)

  • Method Details

    • builder

      public static DefaultRequestHandler.Builder builder()

    • onGetTask

      public org.a2aproject.sdk.spec.Task onGetTask(org.a2aproject.sdk.spec.TaskQueryParams params, ServerCallContext context) throws org.a2aproject.sdk.spec.A2AError
      Specified by:
      onGetTask in interface RequestHandler
      Throws:
      org.a2aproject.sdk.spec.A2AError

    • onListTasks

      public org.a2aproject.sdk.jsonrpc.common.wrappers.ListTasksResult onListTasks(org.a2aproject.sdk.spec.ListTasksParams params, ServerCallContext context) throws org.a2aproject.sdk.spec.A2AError
      Specified by:
      onListTasks in interface RequestHandler
      Throws:
      org.a2aproject.sdk.spec.A2AError

    • onCancelTask

      public org.a2aproject.sdk.spec.Task onCancelTask(org.a2aproject.sdk.spec.CancelTaskParams params, ServerCallContext context) throws org.a2aproject.sdk.spec.A2AError
      Specified by:
      onCancelTask in interface RequestHandler
      Throws:
      org.a2aproject.sdk.spec.A2AError

    • onMessageSend

      public org.a2aproject.sdk.spec.EventKind onMessageSend(org.a2aproject.sdk.spec.MessageSendParams params, ServerCallContext context) throws org.a2aproject.sdk.spec.A2AError
      Specified by:
      onMessageSend in interface RequestHandler
      Throws:
      org.a2aproject.sdk.spec.A2AError

    • onMessageSendStream

      public Flow.Publisher<org.a2aproject.sdk.spec.StreamingEventKind> onMessageSendStream(org.a2aproject.sdk.spec.MessageSendParams params, ServerCallContext context) throws org.a2aproject.sdk.spec.A2AError
      Specified by:
      onMessageSendStream in interface RequestHandler
      Throws:
      org.a2aproject.sdk.spec.A2AError

    • onCreateTaskPushNotificationConfig

      public org.a2aproject.sdk.spec.TaskPushNotificationConfig onCreateTaskPushNotificationConfig(org.a2aproject.sdk.spec.TaskPushNotificationConfig params, ServerCallContext context) throws org.a2aproject.sdk.spec.A2AError
      Specified by:
      onCreateTaskPushNotificationConfig in interface RequestHandler
      Throws:
      org.a2aproject.sdk.spec.A2AError

    • onGetTaskPushNotificationConfig

      public org.a2aproject.sdk.spec.TaskPushNotificationConfig onGetTaskPushNotificationConfig(org.a2aproject.sdk.spec.GetTaskPushNotificationConfigParams params, ServerCallContext context) throws org.a2aproject.sdk.spec.A2AError
      Specified by:
      onGetTaskPushNotificationConfig in interface RequestHandler
      Throws:
      org.a2aproject.sdk.spec.A2AError

    • onSubscribeToTask

      public Flow.Publisher<org.a2aproject.sdk.spec.StreamingEventKind> onSubscribeToTask(org.a2aproject.sdk.spec.TaskIdParams params, ServerCallContext context) throws org.a2aproject.sdk.spec.A2AError
      Specified by:
      onSubscribeToTask in interface RequestHandler
      Throws:
      org.a2aproject.sdk.spec.A2AError

    • onListTaskPushNotificationConfigs

      public org.a2aproject.sdk.spec.ListTaskPushNotificationConfigsResult onListTaskPushNotificationConfigs(org.a2aproject.sdk.spec.ListTaskPushNotificationConfigsParams params, ServerCallContext context) throws org.a2aproject.sdk.spec.A2AError
      Specified by:
      onListTaskPushNotificationConfigs in interface RequestHandler
      Throws:
      org.a2aproject.sdk.spec.A2AError

    • onDeleteTaskPushNotificationConfig

      public void onDeleteTaskPushNotificationConfig(org.a2aproject.sdk.spec.DeleteTaskPushNotificationConfigParams params, ServerCallContext context)
      Specified by:
      onDeleteTaskPushNotificationConfig in interface RequestHandler

    • authorizeTaskAccess

      public void authorizeTaskAccess(@Nullable String requestedTaskId, ServerCallContext context, TaskOperation operation) throws org.a2aproject.sdk.spec.A2AError
      The authorization is done by the AuthorizationRequestHandlerDecorator
      Specified by:
      authorizeTaskAccess in interface RequestHandler
      Throws:
      org.a2aproject.sdk.spec.A2AError