Class GrpcHandler

java.lang.Object
org.a2aproject.sdk.grpc.A2AServiceGrpc.A2AServiceImplBase
org.a2aproject.sdk.transport.grpc.handler.GrpcHandler
All Implemented Interfaces:
io.grpc.BindableService, A2AServiceGrpc.AsyncService
Direct Known Subclasses:
QuarkusGrpcHandler
@Vetoed public abstract class GrpcHandler extends A2AServiceGrpc.A2AServiceImplBase
gRPC transport handler for processing A2A protocol requests.

This abstract class implements the gRPC service interface for the A2A protocol, handling both unary (blocking) and server streaming RPC calls. It translates gRPC requests to A2A protocol operations, coordinates with the request handler and agent executor, and manages error handling with appropriate gRPC status codes.

Request Flow

gRPC Request → GrpcHandler (this class)
    ↓
Protobuf → Domain conversion
    ↓
RequestHandler → AgentExecutor
    ↓
Domain → Protobuf conversion
    ↓
gRPC Response (unary or streaming)

Supported Operations

Unary RPC (blocking):

Server Streaming RPC:

Error Handling

A2A errors are mapped to gRPC status codes:

Context Access

The handler provides rich context information equivalent to Python's grpc.aio.ServicerContext:

Extension Points

Subclasses must implement:

CDI Integration

This class is marked with @Vetoed to prevent direct CDI management. Subclasses should be CDI beans (e.g., @GrpcService in Quarkus) that inject dependencies and provide them through the abstract methods.

See Also:

  • Constructor Details

    • GrpcHandler

      public GrpcHandler()
      Constructs a new GrpcHandler.

  • Method Details

    • sendMessage

      public void sendMessage(SendMessageRequest request, io.grpc.stub.StreamObserver<SendMessageResponse> responseObserver)
      Handles a unary (blocking) message send request.

      This method processes a message send request, waits for the agent to complete processing, and returns either a Task or Message in the response.

      Protocol Flow:

      1. Validate A2A protocol version and extensions
      2. Convert protobuf request to domain MessageSendParams
      3. Invoke RequestHandler.onMessageSend(MessageSendParams, ServerCallContext)
      4. Convert domain response to protobuf SendMessageResponse
      5. Send response and complete the RPC

      Error Handling:

      Parameters:
      request - the gRPC message send request
      responseObserver - the gRPC response stream observer

    • getTask

      public void getTask(GetTaskRequest request, io.grpc.stub.StreamObserver<Task> responseObserver)
      Description copied from interface: A2AServiceGrpc.AsyncService
      Gets the latest state of a task.

    • listTasks

      public void listTasks(ListTasksRequest request, io.grpc.stub.StreamObserver<ListTasksResponse> responseObserver)
      Description copied from interface: A2AServiceGrpc.AsyncService
      Lists tasks that match the specified filter.

    • cancelTask

      public void cancelTask(CancelTaskRequest request, io.grpc.stub.StreamObserver<Task> responseObserver)
      Description copied from interface: A2AServiceGrpc.AsyncService
      Cancels a task in progress.

    • createTaskPushNotificationConfig

      public void createTaskPushNotificationConfig(TaskPushNotificationConfig request, io.grpc.stub.StreamObserver<TaskPushNotificationConfig> responseObserver)
      Description copied from interface: A2AServiceGrpc.AsyncService
      (-- api-linter: client-libraries::4232::required-fields=disabled
    api-linter: core::0133::method-signature=disabled
    api-linter: core::0133::request-message-name=disabled
    aip.dev/not-precedent: method_signature preserved for backwards compatibility --)
Creates a push notification config for a task.

    • getTaskPushNotificationConfig

      public void getTaskPushNotificationConfig(GetTaskPushNotificationConfigRequest request, io.grpc.stub.StreamObserver<TaskPushNotificationConfig> responseObserver)
      Description copied from interface: A2AServiceGrpc.AsyncService
      Gets a push notification config for a task.

    • listTaskPushNotificationConfigs

      public void listTaskPushNotificationConfigs(ListTaskPushNotificationConfigsRequest request, io.grpc.stub.StreamObserver<ListTaskPushNotificationConfigsResponse> responseObserver)
      Description copied from interface: A2AServiceGrpc.AsyncService
      Get a list of push notifications configured for a task.

    • sendStreamingMessage

      public void sendStreamingMessage(SendMessageRequest request, io.grpc.stub.StreamObserver<StreamResponse> responseObserver)
      Handles a server streaming message send request.

      This method processes a message send request with streaming response, where the agent can emit multiple events (artifacts, status updates, messages) as the task progresses.

      Protocol Flow:

      1. Verify streaming capability is enabled in agent card
      2. Validate A2A protocol version and extensions
      3. Convert protobuf request to domain MessageSendParams
      4. Invoke RequestHandler.onMessageSendStream(MessageSendParams, ServerCallContext)
      5. Subscribe to event publisher and stream responses
      6. Convert each domain event to protobuf StreamResponse
      7. Complete RPC when final event received or error occurs

      Streaming Characteristics:

      • Server streaming RPC - server sends multiple responses
      • Backpressure handled through reactive streams subscription
      • Client disconnect detection via gRPC context cancellation
      • Automatic cleanup when stream completes or errors

      Error Handling:

      Parameters:
      request - the gRPC message send request
      responseObserver - the gRPC response stream observer

    • subscribeToTask

      public void subscribeToTask(SubscribeToTaskRequest request, io.grpc.stub.StreamObserver<StreamResponse> responseObserver)
      Description copied from interface: A2AServiceGrpc.AsyncService
      Subscribes to task updates for tasks not in a terminal state.
Returns `UnsupportedOperationError` if the task is already in a terminal state (completed, failed, canceled, rejected).

    • getExtendedAgentCard

      public void getExtendedAgentCard(GetExtendedAgentCardRequest request, io.grpc.stub.StreamObserver<AgentCard> responseObserver)
      Description copied from interface: A2AServiceGrpc.AsyncService
      Gets the extended agent card for the authenticated agent.

    • deleteTaskPushNotificationConfig

      public void deleteTaskPushNotificationConfig(DeleteTaskPushNotificationConfigRequest request, io.grpc.stub.StreamObserver<com.google.protobuf.Empty> responseObserver)
      Description copied from interface: A2AServiceGrpc.AsyncService
      Deletes a push notification config for a task.

    • getDeploymentClassLoader

      protected ClassLoader getDeploymentClassLoader()
      Returns the deployment classloader for this handler.

      Used for transport configuration validation with proper classloader context.

      Returns:
      the deployment classloader

    • setStreamingSubscribedRunnable

      public static void setStreamingSubscribedRunnable(Runnable runnable)
      Sets a callback to be invoked when streaming subscription starts.

      This is a testing hook used to synchronize test execution with streaming setup. In production, this remains null.

      Parameters:
      runnable - the callback to invoke on subscription

    • getRequestHandler

      protected abstract RequestHandler getRequestHandler()
      Returns the request handler instance for processing A2A protocol requests.
      Returns:
      the request handler

    • getAgentCard

      protected abstract @Nullable AgentCard getAgentCard()
      Returns the public agent card defining the agent's capabilities and metadata, or null if no public agent card is configured.
      Returns:
      the agent card, or null if not configured

    • getExtendedAgentCard

      protected abstract AgentCard getExtendedAgentCard()
      Returns the extended agent card with additional capabilities, or null if not configured.
      Returns:
      the extended agent card, or null if not available

    • getCallContextFactory

      protected abstract CallContextFactory getCallContextFactory()
      Returns the custom call context factory, or null to use default context creation.
      Returns:
      the call context factory, or null for default behavior

    • getExecutor

      protected abstract Executor getExecutor()
      Returns the executor for running async operations (streaming subscriptions, etc.).
      Returns:
      the executor

    • getAgentCardRouter

      protected @Nullable AgentCardRouter getAgentCardRouter()
      Returns the optional agent card router for tenant-based resolution, or null. Subclasses may override to provide a router implementation.
      Returns:
      the agent card router, or null if multitenancy is not configured

    • getCurrentMetadata

      protected static @Nullable io.grpc.Metadata getCurrentMetadata()
      Gets the complete gRPC metadata from the current context. Equivalent to Python's context.invocation_metadata.
      Returns:
      the gRPC Metadata object, or null if not available

    • getCurrentMethodName

      protected static @Nullable String getCurrentMethodName()
      Gets the current gRPC method name. Equivalent to Python's context.method().
      Returns:
      the method name, or null if not available

    • getCurrentPeerInfo

      protected static @Nullable String getCurrentPeerInfo()
      Gets the peer information for the current gRPC call. Equivalent to Python's context.peer().
      Returns:
      the peer information, or null if not available