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 and a2a.blocking.consumption.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: