errorConsumer = mock(Consumer.class);
+ A2AClientException expectedException = new A2AClientException("Streaming failed");
+ doThrow(expectedException).when(delegate).sendMessageStreaming(any(MessageSendParams.class), any(Consumer.class),
+ any(Consumer.class), any(ClientCallContext.class));
+
+ A2AClientException exception = assertThrows(A2AClientException.class,
+ () -> transport.sendMessageStreaming(request, eventConsumer, errorConsumer, context));
+
+ assertEquals(expectedException, exception);
+ verify(span).setStatus(StatusCode.ERROR, "Streaming failed");
+ verify(span).end();
+ }
+}
diff --git a/extras/opentelemetry/common/pom.xml b/extras/opentelemetry/common/pom.xml
new file mode 100644
index 000000000..d21655202
--- /dev/null
+++ b/extras/opentelemetry/common/pom.xml
@@ -0,0 +1,18 @@
+
+
+ 4.0.0
+
+
+ io.github.a2asdk
+ a2a-java-sdk-opentelemetry-parent
+ 1.0.0.Alpha2-SNAPSHOT
+
+
+ a2a-java-sdk-opentelemetry-common
+
+ A2A Java SDK :: Extras :: Opentelemetry :: Common
+ Common OpenTelemetry utilities for A2A Java SDK
+
+
diff --git a/extras/opentelemetry/common/src/main/java/io/a2a/extras/opentelemetry/A2AObservabilityNames.java b/extras/opentelemetry/common/src/main/java/io/a2a/extras/opentelemetry/A2AObservabilityNames.java
new file mode 100644
index 000000000..c2821d8a7
--- /dev/null
+++ b/extras/opentelemetry/common/src/main/java/io/a2a/extras/opentelemetry/A2AObservabilityNames.java
@@ -0,0 +1,23 @@
+package io.a2a.extras.opentelemetry;
+
+public interface A2AObservabilityNames {
+
+ String EXTRACT_REQUEST_SYS_PROPERTY = "io.a2a.server.extract.request";
+ String EXTRACT_RESPONSE_SYS_PROPERTY = "io.a2a.server.extract.response";
+
+ String ERROR_TYPE = "error.type";
+
+ String GENAI_PREFIX = "gen_ai.agent.a2a";
+ String GENAI_CONFIG_ID = GENAI_PREFIX + ".config_id";
+ String GENAI_CONTEXT_ID = GENAI_PREFIX + ".context_id"; //gen_ai.conversation.id
+ String GENAI_EXTENSIONS = GENAI_PREFIX + ".extensions";
+ String GENAI_MESSAGE_ID = GENAI_PREFIX + ".message_id";
+ String GENAI_OPERATION_NAME = GENAI_PREFIX + ".operation.name"; //gen_ai.agent.operation.name ?
+ String GENAI_PARTS_NUMBER = GENAI_PREFIX + ".parts.number";
+ String GENAI_PROTOCOL = GENAI_PREFIX + ".protocol";
+ String GENAI_STATUS = GENAI_PREFIX + ".status";
+ String GENAI_REQUEST = GENAI_PREFIX + ".request"; //gen_ai.input.messages ?
+ String GENAI_RESPONSE = GENAI_PREFIX + ".response"; // gen_ai.output.messages ?
+ String GENAI_ROLE = GENAI_PREFIX + ".role";
+ String GENAI_TASK_ID = GENAI_PREFIX + ".task_id";
+}
diff --git a/extras/opentelemetry/common/src/main/java/io/a2a/extras/opentelemetry/package-info.java b/extras/opentelemetry/common/src/main/java/io/a2a/extras/opentelemetry/package-info.java
new file mode 100644
index 000000000..8e7125631
--- /dev/null
+++ b/extras/opentelemetry/common/src/main/java/io/a2a/extras/opentelemetry/package-info.java
@@ -0,0 +1,8 @@
+/**
+ * Common OpenTelemetry utilities and shared components for A2A Java SDK.
+ *
+ * This package contains common utilities, constants, and helper classes
+ * used across both client and server OpenTelemetry integrations.
+ */
+@org.jspecify.annotations.NullMarked
+package io.a2a.extras.opentelemetry;
diff --git a/extras/opentelemetry/integration-tests/README.md b/extras/opentelemetry/integration-tests/README.md
new file mode 100644
index 000000000..a66fe8f6c
--- /dev/null
+++ b/extras/opentelemetry/integration-tests/README.md
@@ -0,0 +1,160 @@
+# OpenTelemetry Integration Tests (Quarkus-based)
+
+## Overview
+
+This module provides **Quarkus-based integration tests** for OpenTelemetry tracing in the A2A Java SDK, similar to the approach used in the [Quarkus OpenTelemetry quickstart](https://github.com/quarkusio/quarkus/tree/main/integration-tests/opentelemetry-quickstart).
+
+Unlike the previous mock-based tests, these are **real integration tests** that:
+- Start an actual Quarkus application
+- Expose a REST API with A2A agent endpoints
+- Make real HTTP requests
+- Validate that OpenTelemetry spans are created correctly
+
+## Architecture
+
+### Components
+
+1. **SimpleAgent** - A basic A2A agent implementation for testing
+ - Implements all RequestHandler methods
+ - Stores tasks in memory
+ - Provides simple echo responses for messages
+
+2. **AgentResource** - JAX-RS REST resource
+ - Exposes HTTP endpoints (`/a2a/tasks`, `/a2a/messages`, etc.)
+ - Delegates to the RequestHandler
+ - Creates ServerCallContext for each request
+
+3. **InstrumentedRequestHandler** - CDI Alternative
+ - Wraps SimpleAgent with OpenTelemetry decorator
+ - Delegates to the OpenTelemetry decorator for span creation
+ - Ensures spans are created for each operation
+
+4. **OpenTelemetryProducer** - CDI bean producer
+ - Creates the Tracer from Quarkus OpenTelemetry
+ - Produces the OpenTelemetryRequestHandlerDecorator (CDI decorator)
+ - Integrates with Quarkus OpenTelemetry extension
+
+## Test Strategy
+
+### Functional Tests (`OpenTelemetryIntegrationTest`)
+- Use `@QuarkusTest` annotation
+- Make real HTTP requests using REST Assured
+- Verify HTTP responses are correct
+- Ensure the application behaves correctly end-to-end
+
+### Tracing Tests (`OpenTelemetryTracingTest`)
+- Use `InMemorySpanExporter` to capture spans
+- Verify that HTTP requests create OpenTelemetry spans
+- Validate span names, kinds (CLIENT/SERVER), and status codes
+- Check that spans are properly ended
+
+## Current Status
+
+### ✅ Completed
+- Project structure and POM configuration
+- Quarkus dependencies and plugins (including opentelemetry-sdk-testing)
+- SimpleAgentExecutor following reference module pattern
+- TestAgentCardProducer with proper JSONRPC interface
+- InMemorySpanExporter producer for span validation
+- OpenTelemetryIntegrationTest using Client API
+- Tests compile and run (service loader issues resolved)
+
+### 🔨 In Progress / Known Issues
+
+1. **Test Execution Timeouts**
+ - Tests are timing out during message send operations
+ - Error: "Timeout waiting for consumption to complete for task test-task-1"
+ - Likely a configuration issue between client (non-streaming) and server (streaming capable)
+ - Need to investigate JSONRPC transport configuration
+
+2. **Span Validation**
+ - InMemorySpanExporter is configured but needs verification
+ - Some tests are not finding expected spans
+ - May need to configure span processor to route to InMemorySpanExporter
+
+## Running the Tests
+
+### Prerequisites
+```bash
+# Build all A2A SDK modules first
+mvn clean install -DskipTests
+```
+
+### Run Integration Tests
+```bash
+# From the integration-tests directory
+mvn clean verify
+
+# Or from the root
+mvn verify -pl extras/opentelemetry/integration-tests -am
+```
+
+### Run Specific Test
+```bash
+mvn test -Dtest=OpenTelemetryIntegrationTest
+```
+
+## Configuration
+
+### Application Properties
+- `src/main/resources/application.properties` - Runtime configuration
+- `src/test/resources/application.properties` - Test-specific configuration
+
+Key settings:
+```properties
+# OpenTelemetry
+quarkus.otel.sdk.disabled=false
+quarkus.otel.traces.enabled=true
+quarkus.otel.service.name=a2a-opentelemetry-integration-test
+
+# Test mode: use in-memory exporter
+quarkus.otel.traces.exporter=none
+```
+
+### beans.xml
+Located at `src/main/resources/META-INF/beans.xml`:
+- Enables CDI bean discovery
+- Configures alternatives (InstrumentedRequestHandler)
+
+## Next Steps
+
+To complete this integration test module:
+
+1. **Resolve CDI ambiguity**
+ - Add `@Named` or custom qualifier to SimpleAgent
+ - Or exclude DefaultRequestHandler from test classpath
+ - Or use `@Alternative` more effectively
+
+2. **Configure span exporter**
+ - Properly wire InMemorySpanExporter into Quarkus OpenTelemetry
+ - May need custom OTel SDK configuration
+
+3. **Add more test scenarios**
+ - Test error handling and error spans
+ - Test streaming operations
+ - Test context propagation across services
+ - Test span attributes and metadata
+
+4. **Performance testing** (optional)
+ - Measure overhead of OpenTelemetry instrumentation
+ - Verify spans don't impact performance significantly
+
+## Comparison with Quarkus Quickstart
+
+This implementation follows the same patterns as the Quarkus OpenTelemetry quickstart:
+- ✅ Uses `@QuarkusTest` for integration tests
+- ✅ Uses REST Assured for HTTP testing
+- ✅ Integrates with Quarkus OpenTelemetry extension
+- ✅ Uses in-memory span exporter for validation
+- ✅ Tests actual HTTP requests, not mocks
+
+Unlike the quickstart which is a standalone app, this module:
+- Tests A2A SDK-specific functionality
+- Validates OpenTelemetry CDI decorator integration
+- Focuses on A2A protocol operations (tasks, messages, etc.)
+
+## References
+
+- [Quarkus OpenTelemetry Guide](https://quarkus.io/guides/opentelemetry)
+- [Quarkus OpenTelemetry Quickstart](https://github.com/quarkusio/quarkus/tree/main/integration-tests/opentelemetry-quickstart)
+- [OpenTelemetry Java Documentation](https://opentelemetry.io/docs/languages/java/)
diff --git a/extras/opentelemetry/integration-tests/pom.xml b/extras/opentelemetry/integration-tests/pom.xml
new file mode 100644
index 000000000..6d2b4734a
--- /dev/null
+++ b/extras/opentelemetry/integration-tests/pom.xml
@@ -0,0 +1,136 @@
+
+
+ 4.0.0
+
+
+ io.github.a2asdk
+ a2a-java-sdk-opentelemetry-parent
+ 1.0.0.Alpha2-SNAPSHOT
+
+
+ a2a-java-sdk-opentelemetry-integration-tests
+
+ A2A Java SDK :: Extras :: OpenTelemetry :: Integration Tests
+ Quarkus-based integration tests for OpenTelemetry support in A2A Java SDK
+
+
+
+
+ ${project.groupId}
+ a2a-java-sdk-client
+
+
+ ${project.groupId}
+ a2a-java-sdk-reference-jsonrpc
+
+
+ ${project.groupId}
+ a2a-java-sdk-opentelemetry-server
+
+
+ ${project.groupId}
+ a2a-java-sdk-opentelemetry-client
+
+
+
+
+ io.quarkus
+ quarkus-opentelemetry
+
+
+
+
+ io.quarkus
+ quarkus-reactive-routes
+
+
+
+
+ jakarta.enterprise
+ jakarta.enterprise.cdi-api
+ provided
+
+
+
+
+ com.google.code.gson
+ gson
+
+
+
+
+ io.quarkus
+ quarkus-junit5
+ test
+
+
+ io.opentelemetry
+ opentelemetry-sdk-testing
+
+
+ io.rest-assured
+ rest-assured
+ test
+
+
+ org.awaitility
+ awaitility
+ test
+
+
+
+
+
+
+ io.quarkus
+ quarkus-maven-plugin
+ true
+
+
+
+ build
+ generate-code
+ generate-code-tests
+
+
+
+
+ --add-opens=java.base/java.lang=ALL-UNNAMED
+
+
+
+ org.apache.maven.plugins
+ maven-surefire-plugin
+
+
+ org.jboss.logmanager.LogManager
+ ${maven.home}
+
+ --add-opens java.base/java.lang=ALL-UNNAMED
+
+
+
+ org.apache.maven.plugins
+ maven-failsafe-plugin
+
+
+
+ integration-test
+ verify
+
+
+
+
+
+ org.jboss.logmanager.LogManager
+ ${maven.home}
+
+ --add-opens java.base/java.lang=ALL-UNNAMED
+
+
+
+
+
+
diff --git a/extras/opentelemetry/integration-tests/src/main/java/io/a2a/extras/opentelemetry/it/A2ATestRoutes.java b/extras/opentelemetry/integration-tests/src/main/java/io/a2a/extras/opentelemetry/it/A2ATestRoutes.java
new file mode 100644
index 000000000..f6f84f6e1
--- /dev/null
+++ b/extras/opentelemetry/integration-tests/src/main/java/io/a2a/extras/opentelemetry/it/A2ATestRoutes.java
@@ -0,0 +1,228 @@
+package io.a2a.extras.opentelemetry.it;
+
+
+
+import static io.vertx.core.http.HttpHeaders.CONTENT_TYPE;
+
+import com.google.gson.Gson;
+import com.google.gson.GsonBuilder;
+import com.google.gson.JsonArray;
+import com.google.gson.JsonElement;
+import com.google.gson.JsonObject;
+import io.a2a.jsonrpc.common.json.JsonUtil;
+import io.a2a.spec.Task;
+import io.a2a.spec.TaskArtifactUpdateEvent;
+import io.a2a.spec.TaskStatusUpdateEvent;
+import io.opentelemetry.api.trace.Span;
+import io.opentelemetry.api.trace.Tracer;
+import io.opentelemetry.context.Scope;
+import io.opentelemetry.sdk.testing.exporter.InMemorySpanExporter;
+import io.opentelemetry.sdk.trace.data.SpanData;
+import io.quarkus.vertx.web.Body;
+import io.quarkus.vertx.web.Param;
+import io.quarkus.vertx.web.Route;
+import io.vertx.ext.web.RoutingContext;
+import jakarta.enterprise.context.ApplicationScoped;
+import jakarta.enterprise.inject.Produces;
+import jakarta.inject.Inject;
+import jakarta.inject.Singleton;
+import java.util.List;
+import java.util.stream.Collectors;
+
+/**
+ * Test routes for OpenTelemetry integration testing.
+ * Exposes test utilities via REST endpoints.
+ */
+@Singleton
+public class A2ATestRoutes {
+
+ private static final String APPLICATION_JSON = "application/json";
+ private static final String TEXT_PLAIN = "text/plain";
+ private static final Gson gson = new GsonBuilder().create();
+
+ @Inject
+ TestUtilsBean testUtilsBean;
+ @Inject
+ InMemorySpanExporter inMemorySpanExporter;
+
+ @Inject
+ Tracer tracer;
+
+ @Route(path = "/test/task", methods = {Route.HttpMethod.POST}, consumes = {APPLICATION_JSON}, type = Route.HandlerType.BLOCKING)
+ public void saveTask(@Body String body, RoutingContext rc) {
+ try {
+ Task task = JsonUtil.fromJson(body, Task.class);
+ testUtilsBean.saveTask(task);
+ rc.response()
+ .setStatusCode(200)
+ .end();
+ } catch (Throwable t) {
+ errorResponse(t, rc);
+ }
+ }
+
+ @Route(path = "/test/task/:taskId", methods = {Route.HttpMethod.GET}, produces = {APPLICATION_JSON}, type = Route.HandlerType.BLOCKING)
+ public void getTask(@Param String taskId, RoutingContext rc) {
+ try {
+ Task task = testUtilsBean.getTask(taskId);
+ if (task == null) {
+ rc.response()
+ .setStatusCode(404)
+ .end();
+ return;
+ }
+ rc.response()
+ .setStatusCode(200)
+ .putHeader(CONTENT_TYPE, APPLICATION_JSON)
+ .end(JsonUtil.toJson(task));
+
+ } catch (Throwable t) {
+ errorResponse(t, rc);
+ }
+ }
+
+ @Route(path = "/test/task/:taskId", methods = {Route.HttpMethod.DELETE}, type = Route.HandlerType.BLOCKING)
+ public void deleteTask(@Param String taskId, RoutingContext rc) {
+ try {
+ Task task = testUtilsBean.getTask(taskId);
+ if (task == null) {
+ rc.response()
+ .setStatusCode(404)
+ .end();
+ return;
+ }
+ testUtilsBean.deleteTask(taskId);
+ rc.response()
+ .setStatusCode(200)
+ .end();
+ } catch (Throwable t) {
+ errorResponse(t, rc);
+ }
+ }
+
+ @Route(path = "/test/queue/ensure/:taskId", methods = {Route.HttpMethod.POST})
+ public void ensureTaskQueue(@Param String taskId, RoutingContext rc) {
+ try {
+ testUtilsBean.ensureQueue(taskId);
+ rc.response()
+ .setStatusCode(200)
+ .end();
+ } catch (Throwable t) {
+ errorResponse(t, rc);
+ }
+ }
+
+ @Route(path = "/test/queue/enqueueTaskStatusUpdateEvent/:taskId", methods = {Route.HttpMethod.POST})
+ public void enqueueTaskStatusUpdateEvent(@Param String taskId, @Body String body, RoutingContext rc) {
+ try {
+ TaskStatusUpdateEvent event = JsonUtil.fromJson(body, TaskStatusUpdateEvent.class);
+ testUtilsBean.enqueueEvent(taskId, event);
+ rc.response()
+ .setStatusCode(200)
+ .end();
+ } catch (Throwable t) {
+ errorResponse(t, rc);
+ }
+ }
+
+ @Route(path = "/test/queue/enqueueTaskArtifactUpdateEvent/:taskId", methods = {Route.HttpMethod.POST})
+ public void enqueueTaskArtifactUpdateEvent(@Param String taskId, @Body String body, RoutingContext rc) {
+ try {
+ TaskArtifactUpdateEvent event = JsonUtil.fromJson(body, TaskArtifactUpdateEvent.class);
+ testUtilsBean.enqueueEvent(taskId, event);
+ rc.response()
+ .setStatusCode(200)
+ .end();
+ } catch (Throwable t) {
+ errorResponse(t, rc);
+ }
+ }
+
+ @Route(path = "/test/queue/childCount/:taskId", methods = {Route.HttpMethod.GET}, produces = {TEXT_PLAIN})
+ public void getChildQueueCount(@Param String taskId, RoutingContext rc) {
+ int count = testUtilsBean.getChildQueueCount(taskId);
+ rc.response()
+ .setStatusCode(200)
+ .end(String.valueOf(count));
+ }
+
+ @Route(path = "/hello", methods = {Route.HttpMethod.GET}, produces = {TEXT_PLAIN})
+ public void hello(RoutingContext rc) {
+ Span span = tracer.spanBuilder("hello").startSpan();
+ try (Scope scope = span.makeCurrent()) {
+ rc.response()
+ .setStatusCode(200)
+ .putHeader(CONTENT_TYPE, TEXT_PLAIN)
+ .end("Hello from Quarkus REST");
+ } finally {
+ span.end();
+ }
+ }
+
+ @Route(path = "/export", methods = {Route.HttpMethod.GET}, produces = {APPLICATION_JSON})
+ public void exportSpans(@Param String taskId, RoutingContext rc) {
+ List spans = inMemorySpanExporter.getFinishedSpanItems()
+ .stream()
+ .filter(sd -> !sd.getName().contains("export") && !sd.getName().contains("reset"))
+ .collect(Collectors.toList());
+ String json = gson.toJson(serialize(spans));
+ rc.response()
+ .setStatusCode(200)
+ .putHeader(CONTENT_TYPE, APPLICATION_JSON)
+ .end(json);
+ }
+
+ private JsonElement serialize(List spanDatas) {
+ JsonArray spans = new JsonArray(spanDatas.size());
+ for (SpanData spanData : spanDatas) {
+ JsonObject jsonObject = new JsonObject();
+
+ jsonObject.addProperty("spanId", spanData.getSpanId());
+ jsonObject.addProperty("traceId", spanData.getTraceId());
+ jsonObject.addProperty("name", spanData.getName());
+ jsonObject.addProperty("kind", spanData.getKind().name());
+ jsonObject.addProperty("ended", spanData.hasEnded());
+
+ jsonObject.addProperty("parentSpanId", spanData.getParentSpanContext().getSpanId());
+ jsonObject.addProperty("parent_spanId", spanData.getParentSpanContext().getSpanId());
+ jsonObject.addProperty("parent_traceId", spanData.getParentSpanContext().getTraceId());
+ jsonObject.addProperty("parent_remote", spanData.getParentSpanContext().isRemote());
+ jsonObject.addProperty("parent_valid", spanData.getParentSpanContext().isValid());
+
+ spanData.getAttributes().forEach((k, v) -> {
+ jsonObject.addProperty("attr_" + k.getKey(), v.toString());
+ });
+
+ spanData.getResource().getAttributes().forEach((k, v) -> {
+ jsonObject.addProperty("resource_" + k.getKey(), v.toString());
+ });
+ spans.add(jsonObject);
+ }
+
+ return spans;
+ }
+
+ @Route(path = "/reset", methods = {Route.HttpMethod.GET}, produces = {TEXT_PLAIN})
+ public void reset(@Param String taskId, RoutingContext rc) {
+ inMemorySpanExporter.reset();
+ rc.response().setStatusCode(200).end();
+ }
+
+ private void errorResponse(Throwable t, RoutingContext rc) {
+ t.printStackTrace();
+ rc.response()
+ .setStatusCode(500)
+ .putHeader(CONTENT_TYPE, TEXT_PLAIN)
+ .end();
+ }
+
+ @ApplicationScoped
+ static class InMemorySpanExporterProducer {
+
+ @Produces
+ @Singleton
+ InMemorySpanExporter inMemorySpanExporter() {
+ return InMemorySpanExporter.create();
+ }
+ }
+}
diff --git a/extras/opentelemetry/integration-tests/src/main/java/io/a2a/extras/opentelemetry/it/SimpleAgentExecutor.java b/extras/opentelemetry/integration-tests/src/main/java/io/a2a/extras/opentelemetry/it/SimpleAgentExecutor.java
new file mode 100644
index 000000000..6d33dd89c
--- /dev/null
+++ b/extras/opentelemetry/integration-tests/src/main/java/io/a2a/extras/opentelemetry/it/SimpleAgentExecutor.java
@@ -0,0 +1,41 @@
+package io.a2a.extras.opentelemetry.it;
+
+import io.a2a.A2A;
+import io.a2a.server.agentexecution.AgentExecutor;
+import io.a2a.server.agentexecution.RequestContext;
+import io.a2a.server.tasks.AgentEmitter;
+import io.a2a.spec.A2AError;
+import io.a2a.spec.TextPart;
+import jakarta.enterprise.context.ApplicationScoped;
+
+/**
+ * Simple AgentExecutor for integration testing.
+ * Echoes back the user's message and completes the task immediately.
+ */
+@ApplicationScoped
+public class SimpleAgentExecutor implements AgentExecutor {
+
+ @Override
+ public void execute(RequestContext context, AgentEmitter emitter) throws A2AError {
+ // If task doesn't exist, create it
+ if (context.getTask() == null) {
+ emitter.submit();
+ }
+
+ // Get the user's message
+ String userText = context.getMessage().parts().stream()
+ .filter(part -> part instanceof TextPart)
+ .map(part -> ((TextPart) part).text())
+ .findFirst()
+ .orElse("");
+
+ // Echo it back
+ String response = "Echo: " + userText;
+ emitter.complete(A2A.toAgentMessage(response));
+ }
+
+ @Override
+ public void cancel(RequestContext context, AgentEmitter emitter) throws A2AError {
+ emitter.cancel();
+ }
+}
diff --git a/extras/opentelemetry/integration-tests/src/main/java/io/a2a/extras/opentelemetry/it/TestAgentCardProducer.java b/extras/opentelemetry/integration-tests/src/main/java/io/a2a/extras/opentelemetry/it/TestAgentCardProducer.java
new file mode 100644
index 000000000..373169fac
--- /dev/null
+++ b/extras/opentelemetry/integration-tests/src/main/java/io/a2a/extras/opentelemetry/it/TestAgentCardProducer.java
@@ -0,0 +1,47 @@
+package io.a2a.extras.opentelemetry.it;
+
+import io.a2a.server.PublicAgentCard;
+import io.a2a.spec.AgentCapabilities;
+import io.a2a.spec.AgentCard;
+import io.a2a.spec.AgentInterface;
+import io.a2a.spec.AgentSkill;
+import jakarta.enterprise.context.ApplicationScoped;
+import jakarta.enterprise.inject.Produces;
+
+import java.util.Collections;
+import java.util.List;
+
+
+/**
+ * Produces the AgentCard for integration testing.
+ */
+@ApplicationScoped
+public class TestAgentCardProducer {
+
+ @Produces
+ @PublicAgentCard
+ public AgentCard agentCard() {
+ return AgentCard.builder()
+ .name("OpenTelemetry Test Agent")
+ .description("Test agent for OpenTelemetry integration tests")
+ .supportedInterfaces(Collections.singletonList(
+ new AgentInterface("JSONRPC", "http://localhost:8081")
+ ))
+ .version("1.0.0-TEST")
+ .documentationUrl("http://example.com/test")
+ .capabilities(AgentCapabilities.builder()
+ .streaming(true)
+ .pushNotifications(false)
+ .build())
+ .defaultInputModes(Collections.singletonList("text"))
+ .defaultOutputModes(Collections.singletonList("text"))
+ .skills(Collections.singletonList(AgentSkill.builder()
+ .id("echo")
+ .name("Echo")
+ .description("Echoes back the user's message")
+ .tags(Collections.singletonList("test"))
+ .examples(List.of("hello", "test message"))
+ .build()))
+ .build();
+ }
+}
diff --git a/extras/opentelemetry/integration-tests/src/main/java/io/a2a/extras/opentelemetry/it/TestUtilsBean.java b/extras/opentelemetry/integration-tests/src/main/java/io/a2a/extras/opentelemetry/it/TestUtilsBean.java
new file mode 100644
index 000000000..594063171
--- /dev/null
+++ b/extras/opentelemetry/integration-tests/src/main/java/io/a2a/extras/opentelemetry/it/TestUtilsBean.java
@@ -0,0 +1,46 @@
+package io.a2a.extras.opentelemetry.it;
+
+import io.a2a.server.events.QueueManager;
+import io.a2a.server.tasks.TaskStore;
+import io.a2a.spec.Event;
+import io.a2a.spec.Task;
+import jakarta.enterprise.context.ApplicationScoped;
+import jakarta.inject.Inject;
+
+/**
+ * Test utilities for OpenTelemetry integration tests.
+ * Allows direct manipulation of tasks and queues for testing.
+ */
+@ApplicationScoped
+public class TestUtilsBean {
+
+ @Inject
+ TaskStore taskStore;
+
+ @Inject
+ QueueManager queueManager;
+
+ public void saveTask(Task task) {
+ taskStore.save(task, false);
+ }
+
+ public Task getTask(String taskId) {
+ return taskStore.get(taskId);
+ }
+
+ public void deleteTask(String taskId) {
+ taskStore.delete(taskId);
+ }
+
+ public void ensureQueue(String taskId) {
+ queueManager.createOrTap(taskId);
+ }
+
+ public void enqueueEvent(String taskId, Event event) {
+ queueManager.get(taskId).enqueueEvent(event);
+ }
+
+ public int getChildQueueCount(String taskId) {
+ return queueManager.getActiveChildQueueCount(taskId);
+ }
+}
diff --git a/extras/opentelemetry/integration-tests/src/main/resources/META-INF/beans.xml b/extras/opentelemetry/integration-tests/src/main/resources/META-INF/beans.xml
new file mode 100644
index 000000000..5badbc80d
--- /dev/null
+++ b/extras/opentelemetry/integration-tests/src/main/resources/META-INF/beans.xml
@@ -0,0 +1,9 @@
+
+
+
+ io.a2a.extras.opentelemetry.it.InstrumentedRequestHandler
+
+
diff --git a/extras/opentelemetry/integration-tests/src/main/resources/application.properties b/extras/opentelemetry/integration-tests/src/main/resources/application.properties
new file mode 100644
index 000000000..ca0d2c4fd
--- /dev/null
+++ b/extras/opentelemetry/integration-tests/src/main/resources/application.properties
@@ -0,0 +1,28 @@
+# Quarkus configuration
+quarkus.application.name=a2a-otel-test
+
+# HTTP configuration
+quarkus.http.port=8081
+quarkus.http.test-port=8081
+
+# OpenTelemetry configuration
+quarkus.otel.sdk.disabled=false
+quarkus.otel.traces.enabled=true
+quarkus.otel.metrics.enabled=false
+quarkus.otel.logs.enabled=false
+
+quarkus.otel.instrument.vertx-http=false
+
+quarkus.otel.bsp.schedule.delay=0
+quarkus.otel.bsp.export.timeout=5s
+
+# Service name
+quarkus.otel.service.name=a2a-opentelemetry-integration-test
+
+# Propagators
+quarkus.otel.propagators=tracecontext
+
+# Logging
+quarkus.log.level=INFO
+quarkus.log.category."io.a2a".level=DEBUG
+quarkus.log.category."io.opentelemetry".level=DEBUG
diff --git a/extras/opentelemetry/integration-tests/src/test/java/io/a2a/extras/opentelemetry/it/BaseTest.java b/extras/opentelemetry/integration-tests/src/test/java/io/a2a/extras/opentelemetry/it/BaseTest.java
new file mode 100644
index 000000000..69304e10e
--- /dev/null
+++ b/extras/opentelemetry/integration-tests/src/test/java/io/a2a/extras/opentelemetry/it/BaseTest.java
@@ -0,0 +1,20 @@
+package io.a2a.extras.opentelemetry.it;
+
+import static io.restassured.RestAssured.get;
+
+import java.util.List;
+import java.util.Map;
+
+import io.restassured.common.mapper.TypeRef;
+
+public class BaseTest {
+
+ protected List> getSpans() {
+ return get("/export").body().as(new TypeRef<>() {
+ });
+ }
+
+ protected void buildGlobalTelemetryInstance() {
+ // Do nothing in JVM mode
+ }
+}
\ No newline at end of file
diff --git a/extras/opentelemetry/integration-tests/src/test/java/io/a2a/extras/opentelemetry/it/OpenTelemetryA2ABaseTest.java b/extras/opentelemetry/integration-tests/src/test/java/io/a2a/extras/opentelemetry/it/OpenTelemetryA2ABaseTest.java
new file mode 100644
index 000000000..2171b2fc1
--- /dev/null
+++ b/extras/opentelemetry/integration-tests/src/test/java/io/a2a/extras/opentelemetry/it/OpenTelemetryA2ABaseTest.java
@@ -0,0 +1,301 @@
+package io.a2a.extras.opentelemetry.it;
+
+import static io.quarkus.vertx.web.ReactiveRoutes.APPLICATION_JSON;
+import static io.restassured.RestAssured.given;
+import static java.net.HttpURLConnection.HTTP_OK;
+import static java.util.concurrent.TimeUnit.SECONDS;
+import static org.awaitility.Awaitility.await;
+import static org.junit.jupiter.api.Assertions.*;
+
+import io.a2a.A2A;
+import io.a2a.client.Client;
+import io.a2a.client.config.ClientConfig;
+import io.a2a.client.transport.jsonrpc.JSONRPCTransport;
+import io.a2a.client.transport.jsonrpc.JSONRPCTransportConfigBuilder;
+import io.a2a.jsonrpc.common.json.JsonUtil;
+import io.a2a.spec.*;
+import io.opentelemetry.api.trace.SpanKind;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import java.net.URI;
+import java.net.http.HttpClient;
+import java.net.http.HttpRequest;
+import java.net.http.HttpResponse;
+import java.nio.charset.StandardCharsets;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+
+/**
+ * Abstract base class for OpenTelemetry A2A integration tests.
+ * Contains common test logic shared between unit and integration test modes.
+ */
+abstract class OpenTelemetryA2ABaseTest extends BaseTest {
+
+ protected Client client;
+ protected final int serverPort = 8081;
+
+ @BeforeEach
+ void setUp() throws A2AClientException {
+ ClientConfig clientConfig = new ClientConfig.Builder()
+ .setStreaming(false)
+ .build();
+
+ client = Client.builder(A2A.getAgentCard("http://localhost:" + serverPort))
+ .clientConfig(clientConfig)
+ .withTransport(JSONRPCTransport.class, new JSONRPCTransportConfigBuilder())
+ .build();
+ }
+
+ @BeforeEach
+ void reset() {
+ await().atMost(5, SECONDS).until(() -> {
+ List> spans = getSpans();
+ if (spans.isEmpty()) {
+ return true;
+ } else {
+ given().get("/reset").then().statusCode(HTTP_OK);
+ return false;
+ }
+ });
+ }
+
+ @Test
+ void testGetTaskCreatesSpans() throws Exception {
+ String taskId = "span-test-task-1";
+ String contextId = "span-test-ctx-1";
+
+ Task task = Task.builder()
+ .id(taskId)
+ .contextId(contextId)
+ .status(new TaskStatus(TaskState.COMPLETED))
+ .history(Collections.emptyList())
+ .artifacts(Collections.emptyList())
+ .build();
+
+ saveTaskInTaskStore(task);
+ reset();
+
+ try {
+ Task retrievedTask = client.getTask(new TaskQueryParams(taskId), null);
+
+ assertNotNull(retrievedTask);
+ assertEquals(taskId, retrievedTask.id());
+
+ Thread.sleep(5000);
+
+ List> spans = getSpans();
+ System.out.println("We have created spans " + spans);
+ assertFalse(spans.isEmpty(), "Should have created spans for getTask operation");
+
+ long serverSpanCount = spans.stream()
+ .filter(span -> SpanKind.valueOf((span.get("kind").toString())) == SpanKind.SERVER)
+ .count();
+ assertTrue(serverSpanCount > 0, "Should have at least one SERVER span");
+
+ Map serverSpan = spans.stream()
+ .filter(span -> SpanKind.valueOf(span.get("kind").toString()) == SpanKind.SERVER)
+ .filter(span -> "GetTask".equals(span.get("name")))
+ .findFirst()
+ .orElseThrow(() -> new AssertionError("No SERVER span found for GetTask"));
+
+ assertEquals("GetTask", serverSpan.get("attr_gen_ai.agent.a2a.operation.name"),
+ "Operation name attribute should be set");
+ assertEquals(taskId, serverSpan.get("attr_gen_ai.agent.a2a.task_id"),
+ "Task ID attribute should be set");
+
+ } finally {
+ deleteTaskInTaskStore(taskId);
+ }
+ }
+
+ @Test
+ void testListTasksCreatesSpans() throws Exception {
+ reset();
+
+ ListTasksParams params = new ListTasksParams(
+ null, null, null, null, null, null, null, ""
+ );
+
+ io.a2a.jsonrpc.common.wrappers.ListTasksResult result = client.listTasks(params, null);
+
+ assertNotNull(result);
+
+ Thread.sleep(5000);
+
+ List> spans = getSpans();
+ System.out.println("We have created spans " + spans);
+ assertFalse(spans.isEmpty(), "Should have created spans for listTasks");
+
+ Map serverSpan = spans.stream()
+ .filter(span -> SpanKind.valueOf(span.get("kind").toString()) == SpanKind.SERVER)
+ .filter(span -> "ListTasks".equals(span.get("name")))
+ .findFirst()
+ .orElseThrow(() -> new AssertionError("No SERVER span found for ListTasks"));
+
+ assertEquals("ListTasks", serverSpan.get("attr_gen_ai.agent.a2a.operation.name"),
+ "Operation name attribute should be set");
+ }
+
+ @Test
+ void testCancelTaskCreatesSpans() throws Exception {
+ String taskId = "cancel-test-task-1";
+ String contextId = "cancel-test-ctx-1";
+
+ Task task = Task.builder()
+ .id(taskId)
+ .contextId(contextId)
+ .status(new TaskStatus(TaskState.WORKING))
+ .history(Collections.emptyList())
+ .artifacts(Collections.emptyList())
+ .build();
+
+ saveTaskInTaskStore(task);
+ ensureQueueForTask(taskId);
+ reset();
+
+ try {
+ Task cancelledTask = client.cancelTask(new TaskIdParams(taskId), null);
+
+ assertNotNull(cancelledTask);
+ assertEquals(TaskState.CANCELED, cancelledTask.status().state());
+
+ Thread.sleep(5000);
+
+ List> spans = getSpans();
+ System.out.println("We have created spans " + spans);
+ assertFalse(spans.isEmpty(), "Should have created spans for cancelTask");
+
+ Map serverSpan = spans.stream()
+ .filter(span -> SpanKind.valueOf(span.get("kind").toString()) == SpanKind.SERVER)
+ .filter(span -> "CancelTask".equals(span.get("name")))
+ .findFirst()
+ .orElseThrow(() -> new AssertionError("No SERVER span found for CancelTask"));
+
+ assertEquals("CancelTask", serverSpan.get("attr_gen_ai.agent.a2a.operation.name"),
+ "Operation name attribute should be set");
+ assertEquals(taskId, serverSpan.get("attr_gen_ai.agent.a2a.task_id"),
+ "Task ID attribute should be set");
+
+ } finally {
+ deleteTaskInTaskStore(taskId);
+ }
+ }
+
+ @Test
+ void testSpanAttributes() throws Exception {
+ String taskId = "attr-test-task-1";
+ String contextId = "attr-test-ctx-1";
+
+ Task task = Task.builder()
+ .id(taskId)
+ .contextId(contextId)
+ .status(new TaskStatus(TaskState.COMPLETED))
+ .history(Collections.emptyList())
+ .artifacts(Collections.emptyList())
+ .build();
+
+ saveTaskInTaskStore(task);
+ reset();
+
+ try {
+ client.getTask(new TaskQueryParams(taskId), null);
+
+ Thread.sleep(5000);
+
+ List> spans = getSpans();
+ System.out.println("We have created spans " + spans);
+ assertFalse(spans.isEmpty());
+
+ Map serverSpan = spans.stream()
+ .filter(span -> SpanKind.valueOf(span.get("kind").toString()) == SpanKind.SERVER)
+ .filter(span -> "GetTask".equals(span.get("name")))
+ .findFirst()
+ .orElseThrow(() -> new AssertionError("No SERVER span found for GetTask"));
+
+ assertNotNull(serverSpan.get("spanId"), "Span should have a span ID");
+ assertNotNull(serverSpan.get("traceId"), "Span should have a trace ID");
+ assertEquals("GetTask", serverSpan.get("name"), "Span name should be GetTask");
+ assertEquals("SERVER", serverSpan.get("kind"), "Span kind should be SERVER");
+ assertEquals(Boolean.TRUE, serverSpan.get("ended"), "Span should be ended");
+
+ assertEquals("GetTask", serverSpan.get("attr_gen_ai.agent.a2a.operation.name"),
+ "Operation name attribute should be set");
+ assertEquals(taskId, serverSpan.get("attr_gen_ai.agent.a2a.task_id"),
+ "Task ID attribute should be set");
+
+ assertNotNull(serverSpan.get("resource_service.name"),
+ "Service name resource attribute should be set");
+ System.out.println("Service name: " + serverSpan.get("resource_service.name"));
+
+ assertNotNull(serverSpan.get("parentSpanId"), "Span should have parent span ID field");
+
+ } finally {
+ deleteTaskInTaskStore(taskId);
+ }
+ }
+
+ protected void saveTaskInTaskStore(Task task) throws Exception {
+ HttpClient httpClient = HttpClient.newBuilder()
+ .version(HttpClient.Version.HTTP_2)
+ .build();
+ HttpRequest request = HttpRequest.newBuilder()
+ .uri(URI.create("http://localhost:" + serverPort + "/test/task"))
+ .POST(HttpRequest.BodyPublishers.ofString(JsonUtil.toJson(task)))
+ .header("Content-Type", APPLICATION_JSON)
+ .build();
+
+ HttpResponse response = httpClient.send(request, HttpResponse.BodyHandlers.ofString(StandardCharsets.UTF_8));
+ if (response.statusCode() != 200) {
+ throw new RuntimeException(String.format("Saving task failed! Status: %d, Body: %s", response.statusCode(), response.body()));
+ }
+ }
+
+ protected Task getTaskFromTaskStore(String taskId) throws Exception {
+ HttpClient httpClient = HttpClient.newBuilder()
+ .version(HttpClient.Version.HTTP_2)
+ .build();
+ HttpRequest request = HttpRequest.newBuilder()
+ .uri(URI.create("http://localhost:" + serverPort + "/test/task/" + taskId))
+ .GET()
+ .build();
+
+ HttpResponse response = httpClient.send(request, HttpResponse.BodyHandlers.ofString(StandardCharsets.UTF_8));
+ if (response.statusCode() == 404) {
+ return null;
+ }
+ if (response.statusCode() != 200) {
+ throw new RuntimeException(String.format("Getting task failed! Status: %d, Body: %s", response.statusCode(), response.body()));
+ }
+ return JsonUtil.fromJson(response.body(), Task.class);
+ }
+
+ protected void deleteTaskInTaskStore(String taskId) throws Exception {
+ HttpClient httpClient = HttpClient.newBuilder()
+ .version(HttpClient.Version.HTTP_2)
+ .build();
+ HttpRequest request = HttpRequest.newBuilder()
+ .uri(URI.create(("http://localhost:" + serverPort + "/test/task/" + taskId)))
+ .DELETE()
+ .build();
+ HttpResponse response = httpClient.send(request, HttpResponse.BodyHandlers.ofString(StandardCharsets.UTF_8));
+ if (response.statusCode() != 200) {
+ throw new RuntimeException(response.statusCode() + ": Deleting task failed!" + response.body());
+ }
+ }
+
+ protected void ensureQueueForTask(String taskId) throws Exception {
+ HttpClient httpClient = HttpClient.newBuilder()
+ .version(HttpClient.Version.HTTP_2)
+ .build();
+ HttpRequest request = HttpRequest.newBuilder()
+ .uri(URI.create("http://localhost:" + serverPort + "/test/queue/ensure/" + taskId))
+ .POST(HttpRequest.BodyPublishers.noBody())
+ .build();
+ HttpResponse response = httpClient.send(request, HttpResponse.BodyHandlers.ofString(StandardCharsets.UTF_8));
+ if (response.statusCode() != 200) {
+ throw new RuntimeException(String.format("Ensuring queue failed! Status: %d, Body: %s", response.statusCode(), response.body()));
+ }
+ }
+}
diff --git a/extras/opentelemetry/integration-tests/src/test/java/io/a2a/extras/opentelemetry/it/OpenTelemetryA2AIT.java b/extras/opentelemetry/integration-tests/src/test/java/io/a2a/extras/opentelemetry/it/OpenTelemetryA2AIT.java
new file mode 100644
index 000000000..7701acd23
--- /dev/null
+++ b/extras/opentelemetry/integration-tests/src/test/java/io/a2a/extras/opentelemetry/it/OpenTelemetryA2AIT.java
@@ -0,0 +1,11 @@
+package io.a2a.extras.opentelemetry.it;
+
+import io.quarkus.test.junit.QuarkusIntegrationTest;
+
+/**
+ * Integration test for OpenTelemetry tracing in A2A SDK.
+ * Runs in native mode with @QuarkusIntegrationTest annotation.
+ */
+@QuarkusIntegrationTest
+class OpenTelemetryA2AIT extends OpenTelemetryA2ABaseTest {
+}
diff --git a/extras/opentelemetry/integration-tests/src/test/java/io/a2a/extras/opentelemetry/it/OpenTelemetryA2ATest.java b/extras/opentelemetry/integration-tests/src/test/java/io/a2a/extras/opentelemetry/it/OpenTelemetryA2ATest.java
new file mode 100644
index 000000000..d6c73b05c
--- /dev/null
+++ b/extras/opentelemetry/integration-tests/src/test/java/io/a2a/extras/opentelemetry/it/OpenTelemetryA2ATest.java
@@ -0,0 +1,11 @@
+package io.a2a.extras.opentelemetry.it;
+
+import io.quarkus.test.junit.QuarkusTest;
+
+/**
+ * Unit test for OpenTelemetry tracing in A2A SDK.
+ * Runs in JVM mode with @QuarkusTest annotation.
+ */
+@QuarkusTest
+class OpenTelemetryA2ATest extends OpenTelemetryA2ABaseTest {
+}
diff --git a/extras/opentelemetry/integration-tests/src/test/java/io/a2a/extras/opentelemetry/it/OpenTelemetryTest.java b/extras/opentelemetry/integration-tests/src/test/java/io/a2a/extras/opentelemetry/it/OpenTelemetryTest.java
new file mode 100644
index 000000000..50011fa81
--- /dev/null
+++ b/extras/opentelemetry/integration-tests/src/test/java/io/a2a/extras/opentelemetry/it/OpenTelemetryTest.java
@@ -0,0 +1,42 @@
+package io.a2a.extras.opentelemetry.it;
+
+import static io.restassured.RestAssured.given;
+import static java.net.HttpURLConnection.HTTP_OK;
+import static java.util.concurrent.TimeUnit.SECONDS;
+import static org.awaitility.Awaitility.await;
+import static org.hamcrest.CoreMatchers.is;
+
+import java.util.List;
+import java.util.Map;
+
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import io.quarkus.test.junit.QuarkusTest;
+
+@QuarkusTest
+public class OpenTelemetryTest extends BaseTest {
+
+ @BeforeEach
+ void reset() {
+ await().atMost(5, SECONDS).until(() -> {
+ List> spans = getSpans();
+ if (spans.isEmpty()) {
+ return true;
+ } else {
+ given().get("/reset").then().statusCode(HTTP_OK);
+ return false;
+ }
+ });
+ }
+
+ @Test
+ void buildTimeEnabled() {
+ given()
+ .when().get("/hello")
+ .then()
+ .statusCode(200)
+ .body(is("Hello from Quarkus REST"));
+ await().atMost(5, SECONDS).until(() -> getSpans().size() == 1);
+ }
+}
\ No newline at end of file
diff --git a/extras/opentelemetry/integration-tests/src/test/resources/application.properties b/extras/opentelemetry/integration-tests/src/test/resources/application.properties
new file mode 100644
index 000000000..1e459ea8e
--- /dev/null
+++ b/extras/opentelemetry/integration-tests/src/test/resources/application.properties
@@ -0,0 +1,25 @@
+# Test configuration
+quarkus.application.name=a2a-otel-test
+
+# HTTP configuration
+quarkus.http.test-port=8081
+
+# OpenTelemetry configuration for testing
+quarkus.otel.sdk.disabled=false
+quarkus.otel.traces.enabled=true
+quarkus.otel.metrics.enabled=false
+quarkus.otel.logs.enabled=false
+quarkus.otel.instrument.vertx-http=false
+
+quarkus.otel.bsp.schedule.delay=0
+quarkus.otel.bsp.export.timeout=5s
+
+# Service name
+quarkus.otel.service.name=a2a-opentelemetry-integration-test
+
+# Propagators
+quarkus.otel.propagators=tracecontext
+
+# Logging
+quarkus.log.level=INFO
+quarkus.log.category."io.a2a".level=DEBUG
diff --git a/extras/opentelemetry/pom.xml b/extras/opentelemetry/pom.xml
new file mode 100644
index 000000000..23e7867de
--- /dev/null
+++ b/extras/opentelemetry/pom.xml
@@ -0,0 +1,41 @@
+
+
+ 4.0.0
+
+
+ io.github.a2asdk
+ a2a-java-sdk-parent
+ 1.0.0.Alpha2-SNAPSHOT
+ ../../pom.xml
+
+
+ a2a-java-sdk-opentelemetry-parent
+ pom
+
+ A2A Java SDK :: Extras :: Opentelemetry :: Parent
+ Java SDK for the Agent2Agent Protocol (A2A) - Extras - Opentelemetry
+
+
+ 2.0.1
+
+
+
+ org.eclipse.microprofile.telemetry
+ microprofile-telemetry-api
+ ${version.org.eclipse.microprofile.telemetry}
+ pom
+ provided
+
+
+
+
+ common
+ client-propagation
+ client
+ server
+ integration-tests
+
+
+
\ No newline at end of file
diff --git a/extras/opentelemetry/server/pom.xml b/extras/opentelemetry/server/pom.xml
new file mode 100644
index 000000000..cef51f092
--- /dev/null
+++ b/extras/opentelemetry/server/pom.xml
@@ -0,0 +1,71 @@
+
+
+ 4.0.0
+
+
+ io.github.a2asdk
+ a2a-java-sdk-opentelemetry-parent
+ 1.0.0.Alpha2-SNAPSHOT
+
+
+ a2a-java-sdk-opentelemetry-server
+
+ A2A Java SDK :: Extras :: Opentelemetry :: Server
+ OpenTelemetry server support for A2A Java SDK
+
+
+
+ ${project.groupId}
+ a2a-java-sdk-opentelemetry-common
+
+
+ ${project.groupId}
+ a2a-java-sdk-server-common
+
+
+ ${project.groupId}
+ a2a-java-sdk-jsonrpc-common
+
+
+ ${project.groupId}
+ a2a-java-sdk-spec-grpc
+
+
+ org.eclipse.microprofile.telemetry
+ microprofile-telemetry-api
+ ${version.org.eclipse.microprofile.telemetry}
+ pom
+ provided
+
+
+ jakarta.enterprise
+ jakarta.enterprise.cdi-api
+
+
+ jakarta.inject
+ jakarta.inject-api
+
+
+ org.slf4j
+ slf4j-api
+
+
+ org.junit.jupiter
+ junit-jupiter
+ test
+
+
+ org.mockito
+ mockito-core
+ test
+
+
+ org.mockito
+ mockito-junit-jupiter
+ test
+
+
+
+
diff --git a/extras/opentelemetry/server/src/main/java/io/a2a/extras/opentelemetry/OpenTelemetryRequestHandlerDecorator.java b/extras/opentelemetry/server/src/main/java/io/a2a/extras/opentelemetry/OpenTelemetryRequestHandlerDecorator.java
new file mode 100644
index 000000000..f8bbf0745
--- /dev/null
+++ b/extras/opentelemetry/server/src/main/java/io/a2a/extras/opentelemetry/OpenTelemetryRequestHandlerDecorator.java
@@ -0,0 +1,465 @@
+package io.a2a.extras.opentelemetry;
+
+import static io.a2a.extras.opentelemetry.A2AObservabilityNames.ERROR_TYPE;
+import static io.a2a.extras.opentelemetry.A2AObservabilityNames.EXTRACT_REQUEST_SYS_PROPERTY;
+import static io.a2a.extras.opentelemetry.A2AObservabilityNames.EXTRACT_RESPONSE_SYS_PROPERTY;
+import static io.a2a.extras.opentelemetry.A2AObservabilityNames.GENAI_CONFIG_ID;
+import static io.a2a.extras.opentelemetry.A2AObservabilityNames.GENAI_CONTEXT_ID;
+import static io.a2a.extras.opentelemetry.A2AObservabilityNames.GENAI_EXTENSIONS;
+import static io.a2a.extras.opentelemetry.A2AObservabilityNames.GENAI_MESSAGE_ID;
+import static io.a2a.extras.opentelemetry.A2AObservabilityNames.GENAI_OPERATION_NAME;
+import static io.a2a.extras.opentelemetry.A2AObservabilityNames.GENAI_PARTS_NUMBER;
+import static io.a2a.extras.opentelemetry.A2AObservabilityNames.GENAI_REQUEST;
+import static io.a2a.extras.opentelemetry.A2AObservabilityNames.GENAI_RESPONSE;
+import static io.a2a.extras.opentelemetry.A2AObservabilityNames.GENAI_ROLE;
+import static io.a2a.extras.opentelemetry.A2AObservabilityNames.GENAI_TASK_ID;
+
+import io.a2a.jsonrpc.common.wrappers.ListTasksResult;
+import io.a2a.server.ServerCallContext;
+import io.a2a.server.requesthandlers.RequestHandler;
+import io.a2a.spec.A2AError;
+import io.a2a.spec.A2AMethods;
+import io.a2a.spec.DeleteTaskPushNotificationConfigParams;
+import io.a2a.spec.EventKind;
+import io.a2a.spec.GetTaskPushNotificationConfigParams;
+import io.a2a.spec.ListTaskPushNotificationConfigParams;
+import io.a2a.spec.ListTaskPushNotificationConfigResult;
+import io.a2a.spec.ListTasksParams;
+import io.a2a.spec.MessageSendParams;
+import io.a2a.spec.StreamingEventKind;
+import io.a2a.spec.Task;
+import io.a2a.spec.TaskIdParams;
+import io.a2a.spec.TaskPushNotificationConfig;
+import io.a2a.spec.TaskQueryParams;
+import io.opentelemetry.api.trace.Span;
+import io.opentelemetry.api.trace.SpanKind;
+import io.opentelemetry.api.trace.StatusCode;
+import io.opentelemetry.api.trace.Tracer;
+import io.opentelemetry.context.Scope;
+import jakarta.annotation.Priority;
+import jakarta.decorator.Decorator;
+import jakarta.decorator.Delegate;
+import jakarta.enterprise.inject.Any;
+import jakarta.inject.Inject;
+import java.util.concurrent.Flow;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * OpenTelemetry CDI Decorator for {@link RequestHandler}.
+ *
+ * This decorator adds distributed tracing to A2A server request handlers.
+ * It creates spans for each request handler method invocation, capturing:
+ *
+ * - Request parameters as span attributes
+ * - Response data as span attributes
+ * - Errors and exceptions with proper status codes
+ * - Timing information for performance monitoring
+ *
+ *
+ * To enable this decorator, add it to your beans.xml:
+ *
{@code
+ *
+ * io.a2a.extras.opentelemetry.OpenTelemetryRequestHandlerDecorator
+ *
+ * }
+ */
+@Decorator
+@Priority(100)
+public abstract class OpenTelemetryRequestHandlerDecorator implements RequestHandler {
+
+ private static final Logger LOGGER = LoggerFactory.getLogger(OpenTelemetryRequestHandlerDecorator.class);
+
+ @Inject
+ @Delegate
+ @Any
+ private RequestHandler delegate;
+
+ @Inject
+ private Tracer tracer;
+
+ /**
+ * Default constructor for CDI.
+ */
+ public OpenTelemetryRequestHandlerDecorator() {
+ }
+
+ /**
+ * Constructor for testing.
+ *
+ * @param delegate the delegate request handler
+ * @param tracer the tracer to use
+ */
+ public OpenTelemetryRequestHandlerDecorator(RequestHandler delegate, Tracer tracer) {
+ this.delegate = delegate;
+ this.tracer = tracer;
+ }
+
+ @Override
+ public Task onGetTask(TaskQueryParams params, ServerCallContext context) throws A2AError {
+ var spanBuilder = tracer.spanBuilder(A2AMethods.GET_TASK_METHOD)
+ .setSpanKind(SpanKind.SERVER)
+ .setAttribute(GENAI_OPERATION_NAME, A2AMethods.GET_TASK_METHOD);
+
+ if (params.id() != null) {
+ spanBuilder.setAttribute(GENAI_TASK_ID, params.id());
+ }
+ if (extractRequest()) {
+ spanBuilder.setAttribute(GENAI_REQUEST, params.toString());
+ }
+
+ Span span = spanBuilder.startSpan();
+
+ try (Scope scope = span.makeCurrent()) {
+ Task result = delegate.onGetTask(params, context);
+
+ if (result != null && extractResponse()) {
+ span.setAttribute(GENAI_RESPONSE, result.toString());
+ }
+
+ span.setStatus(StatusCode.OK);
+ return result;
+ } catch (A2AError error) {
+ span.setAttribute(ERROR_TYPE, error.getMessage());
+ span.setStatus(StatusCode.ERROR, error.getMessage());
+ throw error;
+ } finally {
+ span.end();
+ }
+ }
+
+ @Override
+ public ListTasksResult onListTasks(ListTasksParams params, ServerCallContext context) throws A2AError {
+ var spanBuilder = tracer.spanBuilder(A2AMethods.LIST_TASK_METHOD)
+ .setSpanKind(SpanKind.SERVER)
+ .setAttribute(GENAI_OPERATION_NAME, A2AMethods.LIST_TASK_METHOD);
+
+ if (extractRequest()) {
+ spanBuilder.setAttribute(GENAI_REQUEST, params.toString());
+ }
+ if (params.contextId() != null) {
+ spanBuilder.setAttribute(GENAI_CONTEXT_ID, params.contextId());
+ }
+ Span span = spanBuilder.startSpan();
+
+ try (Scope scope = span.makeCurrent()) {
+ ListTasksResult result = delegate.onListTasks(params, context);
+
+ if (result != null && extractResponse()) {
+ span.setAttribute(GENAI_RESPONSE, result.toString());
+ }
+
+ span.setStatus(StatusCode.OK);
+ return result;
+ } catch (A2AError error) {
+ span.setAttribute(ERROR_TYPE, error.getMessage());
+ span.setStatus(StatusCode.ERROR, error.getMessage());
+ throw error;
+ } finally {
+ span.end();
+ }
+ }
+
+ @Override
+ public Task onCancelTask(TaskIdParams params, ServerCallContext context) throws A2AError {
+ var spanBuilder = tracer.spanBuilder(A2AMethods.CANCEL_TASK_METHOD)
+ .setSpanKind(SpanKind.SERVER)
+ .setAttribute(GENAI_OPERATION_NAME, A2AMethods.CANCEL_TASK_METHOD);
+
+ if (params.id() != null) {
+ spanBuilder.setAttribute(GENAI_TASK_ID, params.id());
+ }
+ if (extractRequest()) {
+ spanBuilder.setAttribute(GENAI_REQUEST, params.toString());
+ }
+
+ Span span = spanBuilder.startSpan();
+
+ try (Scope scope = span.makeCurrent()) {
+ Task result = delegate.onCancelTask(params, context);
+
+ if (result != null && extractResponse()) {
+ span.setAttribute(GENAI_RESPONSE, result.toString());
+ }
+
+ span.setStatus(StatusCode.OK);
+ return result;
+ } catch (A2AError error) {
+ span.setAttribute(ERROR_TYPE, error.getMessage());
+ span.setStatus(StatusCode.ERROR, error.getMessage());
+ throw error;
+ } finally {
+ span.end();
+ }
+ }
+
+ @Override
+ public EventKind onMessageSend(MessageSendParams params, ServerCallContext context) throws A2AError {
+ var spanBuilder = tracer.spanBuilder(A2AMethods.SEND_MESSAGE_METHOD)
+ .setSpanKind(SpanKind.SERVER)
+ .setAttribute(GENAI_OPERATION_NAME, A2AMethods.SEND_MESSAGE_METHOD);
+
+ if (params.message() != null) {
+ if (params.message().taskId() != null) {
+ spanBuilder.setAttribute(GENAI_TASK_ID, params.message().taskId());
+ }
+ if (params.message().contextId() != null) {
+ spanBuilder.setAttribute(GENAI_CONTEXT_ID, params.message().contextId());
+ }
+ if (params.message().messageId() != null) {
+ spanBuilder.setAttribute(GENAI_MESSAGE_ID, params.message().messageId());
+ }
+ if (params.message().role() != null) {
+ spanBuilder.setAttribute(GENAI_ROLE, params.message().role().asString());
+ }
+ if (params.message().extensions() != null && !params.message().extensions().isEmpty()) {
+ spanBuilder.setAttribute(GENAI_EXTENSIONS, String.join(",", params.message().extensions()));
+ }
+ spanBuilder.setAttribute(GENAI_PARTS_NUMBER, params.message().parts().size());
+ }
+ if (extractRequest()) {
+ spanBuilder.setAttribute(GENAI_REQUEST, params.toString());
+ }
+
+ Span span = spanBuilder.startSpan();
+
+ try (Scope scope = span.makeCurrent()) {
+ EventKind result = delegate.onMessageSend(params, context);
+
+ if (result != null && extractResponse()) {
+ span.setAttribute(GENAI_RESPONSE, result.toString());
+ }
+
+ span.setStatus(StatusCode.OK);
+ return result;
+ } catch (A2AError error) {
+ span.setAttribute(ERROR_TYPE, error.getMessage());
+ span.setStatus(StatusCode.ERROR, error.getMessage());
+ throw error;
+ } finally {
+ span.end();
+ }
+ }
+
+ @Override
+ public Flow.Publisher onMessageSendStream(MessageSendParams params, ServerCallContext context) throws A2AError {
+ var spanBuilder = tracer.spanBuilder(A2AMethods.SEND_STREAMING_MESSAGE_METHOD)
+ .setSpanKind(SpanKind.SERVER)
+ .setAttribute(GENAI_OPERATION_NAME, A2AMethods.SEND_STREAMING_MESSAGE_METHOD);
+
+ if (params.message() != null) {
+ if (params.message().taskId() != null) {
+ spanBuilder.setAttribute(GENAI_TASK_ID, params.message().taskId());
+ }
+ if (params.message().contextId() != null) {
+ spanBuilder.setAttribute(GENAI_CONTEXT_ID, params.message().contextId());
+ }
+ if (params.message().messageId() != null) {
+ spanBuilder.setAttribute(GENAI_MESSAGE_ID, params.message().messageId());
+ }
+ if (params.message().role() != null) {
+ spanBuilder.setAttribute(GENAI_ROLE, params.message().role().asString());
+ }
+ if (params.message().extensions() != null && !params.message().extensions().isEmpty()) {
+ spanBuilder.setAttribute(GENAI_EXTENSIONS, String.join(",", params.message().extensions()));
+ }
+ spanBuilder.setAttribute(GENAI_PARTS_NUMBER, params.message().parts().size());
+ }
+ if (extractRequest()) {
+ spanBuilder.setAttribute(GENAI_REQUEST, params.toString());
+ }
+
+ Span span = spanBuilder.startSpan();
+
+ try (Scope scope = span.makeCurrent()) {
+ Flow.Publisher result = delegate.onMessageSendStream(params, context);
+
+ if (extractResponse()) {
+ span.setAttribute(GENAI_RESPONSE, "Stream publisher created");
+ }
+
+ span.setStatus(StatusCode.OK);
+ return result;
+ } catch (A2AError error) {
+ span.setAttribute(ERROR_TYPE, error.getMessage());
+ span.setStatus(StatusCode.ERROR, error.getMessage());
+ throw error;
+ } finally {
+ span.end();
+ }
+ }
+
+ @Override
+ public TaskPushNotificationConfig onCreateTaskPushNotificationConfig(TaskPushNotificationConfig params, ServerCallContext context) throws A2AError {
+ var spanBuilder = tracer.spanBuilder(A2AMethods.SET_TASK_PUSH_NOTIFICATION_CONFIG_METHOD)
+ .setSpanKind(SpanKind.SERVER)
+ .setAttribute(GENAI_OPERATION_NAME, A2AMethods.SET_TASK_PUSH_NOTIFICATION_CONFIG_METHOD);
+
+ if (params.taskId() != null) {
+ spanBuilder.setAttribute(GENAI_TASK_ID, params.taskId());
+ }
+ if (params.pushNotificationConfig() != null && params.pushNotificationConfig().id() != null) {
+ spanBuilder.setAttribute(GENAI_CONFIG_ID, params.pushNotificationConfig().id());
+ }
+ if (extractRequest()) {
+ spanBuilder.setAttribute(GENAI_REQUEST, params.toString());
+ }
+
+ Span span = spanBuilder.startSpan();
+
+ try (Scope scope = span.makeCurrent()) {
+ TaskPushNotificationConfig result = delegate.onCreateTaskPushNotificationConfig(params, context);
+
+ if (result != null && extractResponse()) {
+ span.setAttribute(GENAI_RESPONSE, result.toString());
+ }
+
+ span.setStatus(StatusCode.OK);
+ return result;
+ } catch (A2AError error) {
+ span.setAttribute(ERROR_TYPE, error.getMessage());
+ span.setStatus(StatusCode.ERROR, error.getMessage());
+ throw error;
+ } finally {
+ span.end();
+ }
+ }
+
+ @Override
+ public TaskPushNotificationConfig onGetTaskPushNotificationConfig(GetTaskPushNotificationConfigParams params, ServerCallContext context) throws A2AError {
+ var spanBuilder = tracer.spanBuilder(A2AMethods.GET_TASK_PUSH_NOTIFICATION_CONFIG_METHOD)
+ .setSpanKind(SpanKind.SERVER)
+ .setAttribute(GENAI_OPERATION_NAME, A2AMethods.GET_TASK_PUSH_NOTIFICATION_CONFIG_METHOD);
+
+ if (params.id() != null) {
+ spanBuilder.setAttribute(GENAI_TASK_ID, params.id());
+ }
+ if (params.pushNotificationConfigId() != null) {
+ spanBuilder.setAttribute(GENAI_CONFIG_ID, params.pushNotificationConfigId());
+ }
+ if (extractRequest()) {
+ spanBuilder.setAttribute(GENAI_REQUEST, params.toString());
+ }
+
+ Span span = spanBuilder.startSpan();
+
+ try (Scope scope = span.makeCurrent()) {
+ TaskPushNotificationConfig result = delegate.onGetTaskPushNotificationConfig(params, context);
+
+ if (result != null && extractResponse()) {
+ span.setAttribute(GENAI_RESPONSE, result.toString());
+ }
+
+ span.setStatus(StatusCode.OK);
+ return result;
+ } catch (A2AError error) {
+ span.setAttribute(ERROR_TYPE, error.getMessage());
+ span.setStatus(StatusCode.ERROR, error.getMessage());
+ throw error;
+ } finally {
+ span.end();
+ }
+ }
+
+ @Override
+ public Flow.Publisher onSubscribeToTask(TaskIdParams params, ServerCallContext context) throws A2AError {
+ var spanBuilder = tracer.spanBuilder(A2AMethods.SUBSCRIBE_TO_TASK_METHOD)
+ .setSpanKind(SpanKind.SERVER)
+ .setAttribute(GENAI_OPERATION_NAME, A2AMethods.SUBSCRIBE_TO_TASK_METHOD);
+
+ if (params.id() != null) {
+ spanBuilder.setAttribute(GENAI_TASK_ID, params.id());
+ }
+ if (extractRequest()) {
+ spanBuilder.setAttribute(GENAI_REQUEST, params.toString());
+ }
+
+ Span span = spanBuilder.startSpan();
+
+ try (Scope scope = span.makeCurrent()) {
+ Flow.Publisher result = delegate.onSubscribeToTask(params, context);
+
+ if (extractResponse()) {
+ span.setAttribute(GENAI_RESPONSE, "Stream publisher created");
+ }
+
+ span.setStatus(StatusCode.OK);
+ return result;
+ } catch (A2AError error) {
+ span.setAttribute(ERROR_TYPE, error.getMessage());
+ span.setStatus(StatusCode.ERROR, error.getMessage());
+ throw error;
+ } finally {
+ span.end();
+ }
+ }
+
+ @Override
+ public ListTaskPushNotificationConfigResult onListTaskPushNotificationConfig(ListTaskPushNotificationConfigParams params, ServerCallContext context) throws A2AError {
+ var spanBuilder = tracer.spanBuilder(A2AMethods.LIST_TASK_PUSH_NOTIFICATION_CONFIG_METHOD)
+ .setSpanKind(SpanKind.SERVER)
+ .setAttribute(GENAI_OPERATION_NAME, A2AMethods.LIST_TASK_PUSH_NOTIFICATION_CONFIG_METHOD);
+
+ if (extractRequest()) {
+ spanBuilder.setAttribute(GENAI_REQUEST, params.toString());
+ }
+ if (params.id() != null) {
+ spanBuilder.setAttribute(GENAI_TASK_ID, params.id());
+ }
+
+ Span span = spanBuilder.startSpan();
+
+ try (Scope scope = span.makeCurrent()) {
+ ListTaskPushNotificationConfigResult result = delegate.onListTaskPushNotificationConfig(params, context);
+
+ if (result != null && extractResponse()) {
+ span.setAttribute(GENAI_RESPONSE, result.toString());
+ }
+
+ span.setStatus(StatusCode.OK);
+ return result;
+ } catch (A2AError error) {
+ span.setAttribute(ERROR_TYPE, error.getMessage());
+ span.setStatus(StatusCode.ERROR, error.getMessage());
+ throw error;
+ } finally {
+ span.end();
+ }
+ }
+
+ @Override
+ public void onDeleteTaskPushNotificationConfig(DeleteTaskPushNotificationConfigParams params, ServerCallContext context) throws A2AError {
+ var spanBuilder = tracer.spanBuilder(A2AMethods.DELETE_TASK_PUSH_NOTIFICATION_CONFIG_METHOD)
+ .setSpanKind(SpanKind.SERVER)
+ .setAttribute(GENAI_OPERATION_NAME, A2AMethods.DELETE_TASK_PUSH_NOTIFICATION_CONFIG_METHOD);
+
+ if (extractRequest()) {
+ spanBuilder.setAttribute(GENAI_REQUEST, params.toString());
+ }
+ if (params.id() != null) {
+ spanBuilder.setAttribute(GENAI_TASK_ID, params.id());
+ }
+
+ Span span = spanBuilder.startSpan();
+
+ try (Scope scope = span.makeCurrent()) {
+ delegate.onDeleteTaskPushNotificationConfig(params, context);
+
+ span.setStatus(StatusCode.OK);
+ } catch (A2AError error) {
+ span.setAttribute(ERROR_TYPE, error.getMessage());
+ span.setStatus(StatusCode.ERROR, error.getMessage());
+ throw error;
+ } finally {
+ span.end();
+ }
+ }
+
+ private boolean extractRequest() {
+ return Boolean.getBoolean(EXTRACT_REQUEST_SYS_PROPERTY);
+ }
+
+ private boolean extractResponse() {
+ return Boolean.getBoolean(EXTRACT_RESPONSE_SYS_PROPERTY);
+ }
+}
diff --git a/extras/opentelemetry/server/src/main/java/io/a2a/extras/opentelemetry/package-info.java b/extras/opentelemetry/server/src/main/java/io/a2a/extras/opentelemetry/package-info.java
new file mode 100644
index 000000000..fdd7057d5
--- /dev/null
+++ b/extras/opentelemetry/server/src/main/java/io/a2a/extras/opentelemetry/package-info.java
@@ -0,0 +1,5 @@
+@NullMarked
+package io.a2a.extras.opentelemetry;
+
+import org.jspecify.annotations.NullMarked;
+
diff --git a/extras/opentelemetry/server/src/main/resources/META-INF/beans.xml b/extras/opentelemetry/server/src/main/resources/META-INF/beans.xml
new file mode 100644
index 000000000..dcc4135c8
--- /dev/null
+++ b/extras/opentelemetry/server/src/main/resources/META-INF/beans.xml
@@ -0,0 +1,9 @@
+
+
+
+ io.a2a.extras.opentelemetry.OpenTelemetryRequestHandlerDecorator
+
+
\ No newline at end of file
diff --git a/extras/opentelemetry/server/src/test/java/io/a2a/extras/opentelemetry/OpenTelemetryRequestHandlerDecoratorTest.java b/extras/opentelemetry/server/src/test/java/io/a2a/extras/opentelemetry/OpenTelemetryRequestHandlerDecoratorTest.java
new file mode 100644
index 000000000..2f67e4a05
--- /dev/null
+++ b/extras/opentelemetry/server/src/test/java/io/a2a/extras/opentelemetry/OpenTelemetryRequestHandlerDecoratorTest.java
@@ -0,0 +1,497 @@
+package io.a2a.extras.opentelemetry;
+
+import static io.a2a.extras.opentelemetry.A2AObservabilityNames.ERROR_TYPE;
+import static io.a2a.extras.opentelemetry.A2AObservabilityNames.EXTRACT_REQUEST_SYS_PROPERTY;
+import static io.a2a.extras.opentelemetry.A2AObservabilityNames.EXTRACT_RESPONSE_SYS_PROPERTY;
+import static io.a2a.extras.opentelemetry.A2AObservabilityNames.GENAI_REQUEST;
+import static io.a2a.extras.opentelemetry.A2AObservabilityNames.GENAI_RESPONSE;
+import static org.junit.jupiter.api.Assertions.*;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyString;
+import static org.mockito.ArgumentMatchers.eq;
+import static org.mockito.Mockito.*;
+
+import io.a2a.jsonrpc.common.wrappers.ListTasksResult;
+import io.a2a.server.ServerCallContext;
+import io.a2a.server.requesthandlers.RequestHandler;
+import io.a2a.spec.*;
+import io.opentelemetry.api.trace.Span;
+import io.opentelemetry.api.trace.SpanBuilder;
+import io.opentelemetry.api.trace.SpanKind;
+import io.opentelemetry.api.trace.StatusCode;
+import io.opentelemetry.api.trace.Tracer;
+import io.opentelemetry.context.Scope;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Nested;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.mockito.Mock;
+import org.mockito.junit.jupiter.MockitoExtension;
+
+import java.util.Collections;
+import java.util.List;
+import java.util.concurrent.Flow;
+
+@ExtendWith(MockitoExtension.class)
+class OpenTelemetryRequestHandlerDecoratorTest {
+
+ @Mock
+ private Tracer tracer;
+
+ @Mock
+ private Span span;
+
+ @Mock
+ private SpanBuilder spanBuilder;
+
+ @Mock
+ private Scope scope;
+
+ @Mock
+ private ServerCallContext context;
+
+ @Mock
+ private RequestHandler delegate;
+
+ private TestableOpenTelemetryRequestHandlerDecorator decorator;
+
+ @BeforeEach
+ void setUp() {
+ // Set system properties for extracting request/response
+ System.setProperty(EXTRACT_REQUEST_SYS_PROPERTY, "true");
+ System.setProperty(EXTRACT_RESPONSE_SYS_PROPERTY, "true");
+
+ // Set up the mock chain
+ lenient().when(tracer.spanBuilder(anyString())).thenReturn(spanBuilder);
+ lenient().when(spanBuilder.setSpanKind(any(SpanKind.class))).thenReturn(spanBuilder);
+ lenient().when(spanBuilder.setAttribute(anyString(), anyString())).thenReturn(spanBuilder);
+ lenient().when(spanBuilder.setAttribute(anyString(), anyLong())).thenReturn(spanBuilder);
+ lenient().when(spanBuilder.startSpan()).thenReturn(span);
+ lenient().when(span.makeCurrent()).thenReturn(scope);
+ lenient().when(span.setAttribute(anyString(), anyString())).thenReturn(span);
+ lenient().when(span.setStatus(any(StatusCode.class))).thenReturn(span);
+ lenient().when(span.setStatus(any(StatusCode.class), anyString())).thenReturn(span);
+
+ // Create decorator with mocked dependencies
+ decorator = new TestableOpenTelemetryRequestHandlerDecorator(delegate, tracer);
+ }
+
+ /**
+ * Concrete test implementation of the abstract decorator for testing purposes.
+ */
+ static class TestableOpenTelemetryRequestHandlerDecorator extends OpenTelemetryRequestHandlerDecorator {
+ public TestableOpenTelemetryRequestHandlerDecorator(RequestHandler delegate, Tracer tracer) {
+ super(delegate, tracer);
+ }
+ }
+
+ @Nested
+ class GetTaskTests {
+ @Test
+ void onGetTask_createsSpanAndDelegatesToHandler() throws A2AError {
+ TaskQueryParams params = new TaskQueryParams("task-123", null);
+ Task result = Task.builder()
+ .id("task-123")
+ .contextId("ctx-1")
+ .status(new TaskStatus(TaskState.COMPLETED))
+ .history(Collections.emptyList())
+ .artifacts(Collections.emptyList())
+ .build();
+ when(delegate.onGetTask(params, context)).thenReturn(result);
+
+ Task actualResult = decorator.onGetTask(params, context);
+
+ assertEquals(result, actualResult);
+ verify(tracer).spanBuilder(A2AMethods.GET_TASK_METHOD);
+ verify(spanBuilder).setSpanKind(SpanKind.SERVER);
+ verify(spanBuilder).setAttribute(GENAI_REQUEST, params.toString());
+ verify(spanBuilder).startSpan();
+ verify(span).makeCurrent();
+ verify(span).setAttribute(GENAI_RESPONSE, result.toString());
+ verify(span).setStatus(StatusCode.OK);
+ verify(span).end();
+ verify(delegate).onGetTask(params, context);
+ }
+
+ @Test
+ void onGetTask_withError_setsErrorStatusAndRethrows() throws A2AError {
+ TaskQueryParams params = new TaskQueryParams("task-123", null);
+ A2AError error = new TaskNotFoundError();
+ when(delegate.onGetTask(params, context)).thenThrow(error);
+
+ assertThrows(TaskNotFoundError.class, () -> decorator.onGetTask(params, context));
+
+ verify(span).setAttribute(ERROR_TYPE, error.getMessage());
+ verify(span).setStatus(StatusCode.ERROR, error.getMessage());
+ verify(span).end();
+ }
+ }
+
+ @Nested
+ class ListTasksTests {
+ @Test
+ void onListTasks_createsSpanAndDelegatesToHandler() throws A2AError {
+ ListTasksParams params = new ListTasksParams(null, null, null, null, null, null, null, "test-tenant");
+ ListTasksResult result = new ListTasksResult(Collections.emptyList(), 0, 0, null);
+ when(delegate.onListTasks(params, context)).thenReturn(result);
+
+ ListTasksResult actualResult = decorator.onListTasks(params, context);
+
+ assertEquals(result, actualResult);
+ verify(tracer).spanBuilder(A2AMethods.LIST_TASK_METHOD);
+ verify(spanBuilder).setAttribute(GENAI_REQUEST, params.toString());
+ verify(span).setAttribute(GENAI_RESPONSE, result.toString());
+ verify(span).setStatus(StatusCode.OK);
+ verify(span).end();
+ }
+
+ @Test
+ void onListTasks_withError_setsErrorStatus() throws A2AError {
+ ListTasksParams params = new ListTasksParams(null, null, null, null, null, null, null, "test-tenant");
+ A2AError error = new InvalidRequestError("Invalid parameters");
+ when(delegate.onListTasks(params, context)).thenThrow(error);
+
+ assertThrows(InvalidRequestError.class, () -> decorator.onListTasks(params, context));
+
+ verify(span).setAttribute(ERROR_TYPE, error.getMessage());
+ verify(span).setStatus(StatusCode.ERROR, error.getMessage());
+ verify(span).end();
+ }
+ }
+
+ @Nested
+ class CancelTaskTests {
+ @Test
+ void onCancelTask_createsSpanAndDelegatesToHandler() throws A2AError {
+ TaskIdParams params = new TaskIdParams("task-123");
+ Task result = Task.builder()
+ .id("task-123")
+ .contextId("ctx-1")
+ .status(new TaskStatus(TaskState.CANCELED))
+ .history(Collections.emptyList())
+ .artifacts(Collections.emptyList())
+ .build();
+ when(delegate.onCancelTask(params, context)).thenReturn(result);
+
+ Task actualResult = decorator.onCancelTask(params, context);
+
+ assertEquals(result, actualResult);
+ verify(tracer).spanBuilder(A2AMethods.CANCEL_TASK_METHOD);
+ verify(spanBuilder).setAttribute(GENAI_REQUEST, params.toString());
+ verify(span).setAttribute(GENAI_RESPONSE, result.toString());
+ verify(span).setStatus(StatusCode.OK);
+ verify(span).end();
+ }
+
+ @Test
+ void onCancelTask_withError_setsErrorStatus() throws A2AError {
+ TaskIdParams params = new TaskIdParams("task-123");
+ A2AError error = new TaskNotFoundError();
+ when(delegate.onCancelTask(params, context)).thenThrow(error);
+
+ assertThrows(TaskNotFoundError.class, () -> decorator.onCancelTask(params, context));
+
+ verify(span).setAttribute(ERROR_TYPE, error.getMessage());
+ verify(span).setStatus(StatusCode.ERROR, error.getMessage());
+ verify(span).end();
+ }
+ }
+
+ @Nested
+ class MessageSendTests {
+ @Test
+ void onMessageSend_createsSpanAndDelegatesToHandler() throws A2AError {
+ Message message = Message.builder()
+ .role(Message.Role.USER)
+ .parts(List.of(new TextPart("test message")))
+ .messageId("msg-123")
+ .contextId("ctx-1")
+ .taskId("task-123")
+ .build();
+ MessageSendParams params = new MessageSendParams(message, null, null, "");
+ EventKind result = Task.builder()
+ .id("task-123")
+ .contextId("ctx-1")
+ .status(new TaskStatus(TaskState.COMPLETED))
+ .history(Collections.emptyList())
+ .artifacts(Collections.emptyList())
+ .build();
+ when(delegate.onMessageSend(params, context)).thenReturn(result);
+
+ EventKind actualResult = decorator.onMessageSend(params, context);
+
+ assertEquals(result, actualResult);
+ verify(tracer).spanBuilder(A2AMethods.SEND_MESSAGE_METHOD);
+ verify(spanBuilder).setAttribute(GENAI_REQUEST, params.toString());
+ verify(span).setAttribute(GENAI_RESPONSE, result.toString());
+ verify(span).setStatus(StatusCode.OK);
+ }
+
+ @Test
+ void onMessageSend_withError_setsErrorStatus() throws A2AError {
+ Message message = Message.builder()
+ .role(Message.Role.USER)
+ .parts(List.of(new TextPart("test message")))
+ .messageId("msg-123")
+ .contextId("ctx-1")
+ .taskId("task-123")
+ .build();
+ MessageSendParams params = new MessageSendParams(message, null, null, "");
+ A2AError error = new InvalidRequestError("Invalid message");
+ when(delegate.onMessageSend(params, context)).thenThrow(error);
+
+ assertThrows(InvalidRequestError.class, () -> decorator.onMessageSend(params, context));
+
+ verify(span).setAttribute(ERROR_TYPE, error.getMessage());
+ verify(span).setStatus(StatusCode.ERROR, error.getMessage());
+ verify(span).end();
+ }
+ }
+
+ @Nested
+ class MessageSendStreamTests {
+ @Test
+ void onMessageSendStream_createsSpanWithSpecialMessage() throws A2AError {
+ Message message = Message.builder()
+ .role(Message.Role.USER)
+ .parts(List.of(new TextPart("test message")))
+ .messageId("msg-123")
+ .contextId("ctx-1")
+ .taskId("task-123")
+ .build();
+ MessageSendParams params = new MessageSendParams(message, null, null, "");
+ Flow.Publisher publisher = mock(Flow.Publisher.class);
+ when(delegate.onMessageSendStream(params, context)).thenReturn(publisher);
+
+ Flow.Publisher actualResult = decorator.onMessageSendStream(params, context);
+
+ assertEquals(publisher, actualResult);
+ verify(tracer).spanBuilder(A2AMethods.SEND_STREAMING_MESSAGE_METHOD);
+ verify(spanBuilder).setAttribute(GENAI_REQUEST, params.toString());
+ verify(span).setAttribute(GENAI_RESPONSE, "Stream publisher created");
+ verify(span).setStatus(StatusCode.OK);
+ }
+
+ @Test
+ void onMessageSendStream_withError_setsErrorStatus() throws A2AError {
+ Message message = Message.builder()
+ .role(Message.Role.USER)
+ .parts(List.of(new TextPart("test message")))
+ .messageId("msg-123")
+ .contextId("ctx-1")
+ .taskId("task-123")
+ .build();
+ MessageSendParams params = new MessageSendParams(message, null, null, "");
+ A2AError error = new InvalidRequestError("Stream error");
+ when(delegate.onMessageSendStream(params, context)).thenThrow(error);
+
+ assertThrows(InvalidRequestError.class, () -> decorator.onMessageSendStream(params, context));
+
+ verify(span).setAttribute(ERROR_TYPE, error.getMessage());
+ verify(span).setStatus(StatusCode.ERROR, error.getMessage());
+ verify(span).end();
+ }
+ }
+
+ @Nested
+ class SetTaskPushNotificationConfigTests {
+ @Test
+ void onSetTaskPushNotificationConfig_createsSpanAndDelegatesToHandler() throws A2AError {
+ PushNotificationConfig config = new PushNotificationConfig("http://example.com", null, null, "config-1");
+ TaskPushNotificationConfig params = new TaskPushNotificationConfig("task-123", config, null);
+ TaskPushNotificationConfig result = new TaskPushNotificationConfig("task-123", config, null);
+ when(delegate.onCreateTaskPushNotificationConfig(params, context)).thenReturn(result);
+
+ TaskPushNotificationConfig actualResult = decorator.onCreateTaskPushNotificationConfig(params, context);
+
+ assertEquals(result, actualResult);
+ verify(tracer).spanBuilder(A2AMethods.SET_TASK_PUSH_NOTIFICATION_CONFIG_METHOD);
+ verify(spanBuilder).setAttribute(GENAI_REQUEST, params.toString());
+ verify(span).setAttribute(GENAI_RESPONSE, result.toString());
+ verify(span).setStatus(StatusCode.OK);
+ verify(span).end();
+ }
+
+ @Test
+ void onSetTaskPushNotificationConfig_withError_setsErrorStatus() throws A2AError {
+ PushNotificationConfig config = new PushNotificationConfig("http://example.com", null, null, "config-1");
+ TaskPushNotificationConfig params = new TaskPushNotificationConfig("task-123", config, null);
+ A2AError error = new InvalidRequestError("Invalid config");
+ when(delegate.onCreateTaskPushNotificationConfig(params, context)).thenThrow(error);
+
+ assertThrows(InvalidRequestError.class, () -> decorator.onCreateTaskPushNotificationConfig(params, context));
+
+ verify(span).setAttribute(ERROR_TYPE, error.getMessage());
+ verify(span).setStatus(StatusCode.ERROR, error.getMessage());
+ verify(span).end();
+ }
+ }
+
+ @Nested
+ class GetTaskPushNotificationConfigTests {
+ @Test
+ void onGetTaskPushNotificationConfig_createsSpanAndDelegatesToHandler() throws A2AError {
+ GetTaskPushNotificationConfigParams params = new GetTaskPushNotificationConfigParams("task-123", "config-1");
+ PushNotificationConfig config = new PushNotificationConfig("http://example.com", null, null, "config-1");
+ TaskPushNotificationConfig result = new TaskPushNotificationConfig("task-123", config, null);
+ when(delegate.onGetTaskPushNotificationConfig(params, context)).thenReturn(result);
+
+ TaskPushNotificationConfig actualResult = decorator.onGetTaskPushNotificationConfig(params, context);
+
+ assertEquals(result, actualResult);
+ verify(tracer).spanBuilder(A2AMethods.GET_TASK_PUSH_NOTIFICATION_CONFIG_METHOD);
+ verify(spanBuilder).setAttribute(GENAI_REQUEST, params.toString());
+ verify(span).setAttribute(GENAI_RESPONSE, result.toString());
+ verify(span).setStatus(StatusCode.OK);
+ verify(span).end();
+ }
+
+ @Test
+ void onGetTaskPushNotificationConfig_withError_setsErrorStatus() throws A2AError {
+ GetTaskPushNotificationConfigParams params = new GetTaskPushNotificationConfigParams("task-123", "");
+ A2AError error = new TaskNotFoundError();
+ when(delegate.onGetTaskPushNotificationConfig(params, context)).thenThrow(error);
+
+ assertThrows(TaskNotFoundError.class, () -> decorator.onGetTaskPushNotificationConfig(params, context));
+
+ verify(span).setAttribute(ERROR_TYPE, error.getMessage());
+ verify(span).setStatus(StatusCode.ERROR, error.getMessage());
+ verify(span).end();
+ }
+ }
+
+ @Nested
+ class ResubscribeToTaskTests {
+ @Test
+ void onResubscribeToTask_createsSpanWithSpecialMessage() throws A2AError {
+ TaskIdParams params = new TaskIdParams("task-123");
+ Flow.Publisher publisher = mock(Flow.Publisher.class);
+ when(delegate.onSubscribeToTask(params, context)).thenReturn(publisher);
+
+ Flow.Publisher actualResult = decorator.onSubscribeToTask(params, context);
+
+ assertEquals(publisher, actualResult);
+ verify(tracer).spanBuilder(A2AMethods.SUBSCRIBE_TO_TASK_METHOD);
+ verify(spanBuilder).setAttribute(GENAI_REQUEST, params.toString());
+ verify(span).setAttribute(GENAI_RESPONSE, "Stream publisher created");
+ verify(span).setStatus(StatusCode.OK);
+ }
+
+ @Test
+ void onResubscribeToTask_withError_setsErrorStatus() throws A2AError {
+ TaskIdParams params = new TaskIdParams("task-123");
+ A2AError error = new TaskNotFoundError();
+ when(delegate.onSubscribeToTask(params, context)).thenThrow(error);
+
+ assertThrows(TaskNotFoundError.class, () -> decorator.onSubscribeToTask(params, context));
+
+ verify(span).setAttribute(ERROR_TYPE, error.getMessage());
+ verify(span).setStatus(StatusCode.ERROR, error.getMessage());
+ verify(span).end();
+ }
+ }
+
+ @Nested
+ class ListTaskPushNotificationConfigTests {
+ @Test
+ void onListTaskPushNotificationConfig_createsSpanAndDelegatesToHandler() throws A2AError {
+ ListTaskPushNotificationConfigParams params = new ListTaskPushNotificationConfigParams("task-123");
+ ListTaskPushNotificationConfigResult result = new ListTaskPushNotificationConfigResult(Collections.emptyList(), null);
+ when(delegate.onListTaskPushNotificationConfig(params, context)).thenReturn(result);
+
+ ListTaskPushNotificationConfigResult actualResult = decorator.onListTaskPushNotificationConfig(params, context);
+
+ assertEquals(result, actualResult);
+ verify(tracer).spanBuilder(A2AMethods.LIST_TASK_PUSH_NOTIFICATION_CONFIG_METHOD);
+ verify(spanBuilder).setAttribute(GENAI_REQUEST, params.toString());
+ verify(span).setAttribute(GENAI_RESPONSE, result.toString());
+ verify(span).setStatus(StatusCode.OK);
+ verify(span).end();
+ }
+
+ @Test
+ void onListTaskPushNotificationConfig_withError_setsErrorStatus() throws A2AError {
+ ListTaskPushNotificationConfigParams params = new ListTaskPushNotificationConfigParams("task-123");
+ A2AError error = new InvalidRequestError("Invalid request");
+ when(delegate.onListTaskPushNotificationConfig(params, context)).thenThrow(error);
+
+ assertThrows(InvalidRequestError.class, () -> decorator.onListTaskPushNotificationConfig(params, context));
+
+ verify(span).setAttribute(ERROR_TYPE, error.getMessage());
+ verify(span).setStatus(StatusCode.ERROR, error.getMessage());
+ verify(span).end();
+ }
+ }
+
+ @Nested
+ class DeleteTaskPushNotificationConfigTests {
+ @Test
+ void onDeleteTaskPushNotificationConfig_createsSpanAndDelegatesToHandler() throws A2AError {
+ DeleteTaskPushNotificationConfigParams params = new DeleteTaskPushNotificationConfigParams("task-123", "config-123");
+ doNothing().when(delegate).onDeleteTaskPushNotificationConfig(params, context);
+
+ decorator.onDeleteTaskPushNotificationConfig(params, context);
+
+ verify(tracer).spanBuilder(A2AMethods.DELETE_TASK_PUSH_NOTIFICATION_CONFIG_METHOD);
+ verify(spanBuilder).setAttribute(GENAI_REQUEST, params.toString());
+ verify(span).setStatus(StatusCode.OK);
+ verify(span, never()).setAttribute(eq(GENAI_RESPONSE), anyString());
+ verify(span).end();
+ }
+
+ @Test
+ void onDeleteTaskPushNotificationConfig_withError_setsErrorStatus() throws A2AError {
+ DeleteTaskPushNotificationConfigParams params = new DeleteTaskPushNotificationConfigParams("task-123", "config-123");
+ A2AError error = new TaskNotFoundError();
+ doThrow(error).when(delegate).onDeleteTaskPushNotificationConfig(params, context);
+
+ assertThrows(TaskNotFoundError.class, () -> decorator.onDeleteTaskPushNotificationConfig(params, context));
+
+ verify(span).setAttribute(ERROR_TYPE, error.getMessage());
+ verify(span).setStatus(StatusCode.ERROR, error.getMessage());
+ verify(span).end();
+ }
+ }
+
+ @Nested
+ class SpanLifecycleTests {
+ @Test
+ void allMethods_createAndEndSpans() throws A2AError {
+ TaskQueryParams params = new TaskQueryParams("task-123", null);
+ Task result = Task.builder()
+ .id("task-123")
+ .contextId("ctx-1")
+ .status(new TaskStatus(TaskState.COMPLETED))
+ .history(Collections.emptyList())
+ .artifacts(Collections.emptyList())
+ .build();
+ when(delegate.onGetTask(params, context)).thenReturn(result);
+
+ decorator.onGetTask(params, context);
+
+ verify(span, times(1)).makeCurrent();
+ verify(span, times(1)).end();
+ }
+
+ @Test
+ void spanAttributes_setCorrectly() throws A2AError {
+ TaskQueryParams params = new TaskQueryParams("task-123", null);
+ Task result = Task.builder()
+ .id("task-123")
+ .contextId("ctx-1")
+ .status(new TaskStatus(TaskState.COMPLETED))
+ .history(Collections.emptyList())
+ .artifacts(Collections.emptyList())
+ .build();
+ when(delegate.onGetTask(params, context)).thenReturn(result);
+
+ decorator.onGetTask(params, context);
+
+ verify(tracer).spanBuilder(A2AMethods.GET_TASK_METHOD);
+ verify(spanBuilder).setSpanKind(SpanKind.SERVER);
+ verify(spanBuilder).setAttribute(GENAI_REQUEST, params.toString());
+ verify(span).setAttribute(GENAI_RESPONSE, result.toString());
+ verify(span).setStatus(StatusCode.OK);
+ }
+ }
+}
diff --git a/pom.xml b/pom.xml
index 8f3545dba..dcf90ba44 100644
--- a/pom.xml
+++ b/pom.xml
@@ -189,6 +189,36 @@
a2a-java-sdk-reference-rest
${project.version}