AAPP-1159: Add OTLP traces backend with dual-write to DO Traces API

gradient_adk/__init__.py

@@ -28,4 +28,4 @@
     "add_agent_span",
 ]
 
-__version__ = "0.0.5"
+__version__ = "0.0.8"

gradient_adk/a2a/__init__.py

@@ -1,11 +1,18 @@
 """A2A protocol v0.3.0 integration for Gradient agents.
 
-Public API:
-    create_a2a_server: Create an A2A-enabled FastAPI server for a Gradient agent
+Keep server-only imports lazy so domain-only consumers and tests do not need the
+optional HTTP-server bits of the A2A SDK just to import `gradient_adk.a2a.*`.
 """
 
-from gradient_adk.a2a.infrastructure.server import create_a2a_server
+from __future__ import annotations
 
-__all__ = [
-    "create_a2a_server",
-]
+from typing import Any
+
+
+def create_a2a_server(*args: Any, **kwargs: Any):
+    from gradient_adk.a2a.infrastructure.server import create_a2a_server as _impl
+
+    return _impl(*args, **kwargs)
+
+
+__all__ = ["create_a2a_server"]

gradient_adk/a2a/adapters/primary/a2a_executor.py

@@ -2,6 +2,7 @@
 
 from a2a.server.agent_execution import AgentExecutor, RequestContext
 from a2a.server.events import EventQueue
+from a2a.helpers import new_task_from_user_message
 
 from gradient_adk.a2a.domain.models import GradientOutput, TransformationError
 from gradient_adk.a2a.domain.transformation import MessageTransformer, OutputExtractor
@@ -56,6 +57,9 @@ async def execute(
             and publishes failure event before re-raising to maintain exception
             propagation.
         """
+        if context.current_task is None and context.message is not None:
+            await event_queue.enqueue_event(new_task_from_user_message(context.message))
+
         event_publisher = self.event_publisher_factory(
             event_queue,
             context.context_id,

gradient_adk/a2a/adapters/secondary/event_publisher.py

@@ -3,12 +3,17 @@
 from a2a.server.tasks import TaskUpdater
 from a2a.server.events import EventQueue
 from a2a.types import TaskState
-from a2a.utils import new_agent_text_message
+from a2a.helpers import new_text_message
 
 from gradient_adk.a2a.domain.models import DomainMessage, ValidationError, TransformationError
 from gradient_adk.a2a.domain.ports import EventPort
 
 
+TASK_STATE_COMPLETED = TaskState.Value("TASK_STATE_COMPLETED")
+TASK_STATE_FAILED = TaskState.Value("TASK_STATE_FAILED")
+TASK_STATE_CANCELED = TaskState.Value("TASK_STATE_CANCELED")
+
+
 class A2AEventPublisher:
     """Publishes events to SDK EventQueue. Implements EventPort."""
 
@@ -29,14 +34,14 @@ async def publish_completed(
         message: DomainMessage,
     ) -> None:
         """Publish task completion."""
-        sdk_message = new_agent_text_message(
+        sdk_message = new_text_message(
             text=message.text,
             context_id=self.context_id,
             task_id=task_id,
         )
 
         await self.task_updater.update_status(
-            TaskState.completed,
+            TASK_STATE_COMPLETED,
             message=sdk_message,
         )
 
@@ -47,17 +52,17 @@ async def publish_failed(
     ) -> None:
         """Publish task failure."""
         error_text = f"Agent error: {error.message}"
-        sdk_message = new_agent_text_message(
+        sdk_message = new_text_message(
             text=error_text,
             context_id=self.context_id,
             task_id=task_id,
         )
 
         await self.task_updater.update_status(
-            TaskState.failed,
+            TASK_STATE_FAILED,
             message=sdk_message,
         )
 
     async def publish_canceled(self, task_id: str) -> None:
         """Publish task cancellation."""
-        await self.task_updater.update_status(TaskState.canceled)
+        await self.task_updater.update_status(TASK_STATE_CANCELED)

gradient_adk/a2a/infrastructure/composition.py

@@ -3,6 +3,7 @@
 from typing import Any
 
 from a2a.server.tasks import InMemoryTaskStore
+from a2a.types import AgentCard
 
 from gradient_adk.a2a.domain.validation import MessageValidator
 from gradient_adk.a2a.domain.transformation import MessageTransformer, OutputExtractor
@@ -48,6 +49,7 @@ def event_publisher_factory(event_queue, context_id, task_id):
 
 def compose_request_handler(
     agent_func: Any,
+    agent_card: AgentCard,
     input_key: str = "prompt",
     output_keys: list[str] | None = None,
 ) -> ValidatingRequestHandler:
@@ -78,5 +80,6 @@ def compose_request_handler(
     return ValidatingRequestHandler(
         agent_executor=executor,
         task_store=InMemoryTaskStore(),
+        agent_card=agent_card,
         validator=validator,
     )

gradient_adk/a2a/infrastructure/server.py

@@ -3,9 +3,10 @@
 import os
 from typing import Any
 
-from fastapi import FastAPI
+from fastapi import FastAPI, Request
+from fastapi.responses import JSONResponse
 
-from a2a.server.apps.jsonrpc import A2AFastAPIApplication
+from a2a.server.routes import create_jsonrpc_routes
 from a2a.types import AgentCard, AgentCapabilities
 
 from gradient_adk.a2a.infrastructure.composition import compose_request_handler
@@ -72,30 +73,67 @@ def create_a2a_server(
         name=agent_name,
         version=A2A_PROTOCOL_VERSION,
         description="A2A-enabled Gradient agent",
-        url=base_url,
         capabilities=AgentCapabilities(
             streaming=False,
             push_notifications=False,
-            state_transition_history=True,
         ),
-        supports_authenticated_extended_card=False,
         default_input_modes=["text/plain"],
         default_output_modes=["text/plain"],
         skills=[],
     )
 
     request_handler = compose_request_handler(
         agent_func,
+        agent_card=agent_card,
         input_key=input_key,
         output_keys=output_keys,
     )
 
-    app_builder = A2AFastAPIApplication(
-        agent_card=agent_card,
-        http_handler=request_handler,
-    )
-
-    return app_builder.build(
+    app = FastAPI()
+    jsonrpc_routes = create_jsonrpc_routes(
+        request_handler=request_handler,
         rpc_url=rpc_url,
-        agent_card_url=agent_card_url
+        enable_v0_3_compat=True,
     )
+    jsonrpc_endpoint = jsonrpc_routes[0].endpoint
+
+    @app.post(rpc_url)
+    async def handle_jsonrpc(request: Request):
+        body = await request.json()
+        if body.get("method") == "message/stream":
+            return JSONResponse(
+                {
+                    "jsonrpc": "2.0",
+                    "id": body.get("id"),
+                    "error": {
+                        "code": -32603,
+                        "message": "Streaming is not supported by the agent",
+                    },
+                }
+            )
+        return await jsonrpc_endpoint(request)
+
+    @app.get(agent_card_url)
+    async def get_agent_card():
+        # Serve a compatibility shape that matches the contract exercised by the
+        # repo's A2A tests while sourcing the core values from the SDK proto.
+        return JSONResponse(
+            {
+                "name": agent_card.name,
+                "version": agent_card.version,
+                "description": agent_card.description,
+                "url": base_url,
+                "capabilities": {
+                    "streaming": bool(agent_card.capabilities.streaming),
+                    "pushNotifications": bool(
+                        agent_card.capabilities.push_notifications
+                    ),
+                    "stateTransitionHistory": True,
+                },
+                "defaultInputModes": list(agent_card.default_input_modes),
+                "defaultOutputModes": list(agent_card.default_output_modes),
+                "skills": list(agent_card.skills),
+            }
+        )
+
+    return app

gradient_adk/a2a/infrastructure/validating_handler.py

@@ -9,8 +9,7 @@
 
 from a2a.server.request_handlers import DefaultRequestHandler
 from a2a.server.context import ServerCallContext
-from a2a.types import Message, Task, MessageSendParams, FilePart, DataPart, InvalidParamsError
-from a2a.utils.errors import ServerError
+from a2a.types import AgentCard, Message, Task, SendMessageRequest, InvalidParamsError
 
 from gradient_adk.a2a.domain.validation import MessageValidator
 from gradient_adk.a2a.domain.models import DomainMessage, MessageRole
@@ -32,7 +31,13 @@ class ValidatingRequestHandler(DefaultRequestHandler):
     - Executor can assume input is valid
     """
 
-    def __init__(self, agent_executor: Any, task_store: Any, validator: MessageValidator):
+    def __init__(
+        self,
+        agent_executor: Any,
+        task_store: Any,
+        agent_card: AgentCard,
+        validator: MessageValidator,
+    ):
         """
         Initialize validating request handler.
 
@@ -41,12 +46,12 @@ def __init__(self, agent_executor: Any, task_store: Any, validator: MessageValid
             task_store: The task store for persistence
             validator: Domain validator for business rules
         """
-        super().__init__(agent_executor, task_store)
+        super().__init__(agent_executor, task_store, agent_card)
         self.validator = validator
 
     async def on_message_send(
         self,
-        params: MessageSendParams,
+        params: SendMessageRequest,
         context: ServerCallContext | None = None,
     ) -> Message | Task:
         """
@@ -79,7 +84,7 @@ async def on_message_send(
         if validation_result.is_err():
             error = validation_result.error
             error_message = f"[{error.code}] {error.message}"
-            raise ServerError(InvalidParamsError(message=error_message))
+            raise InvalidParamsError(message=error_message)
 
         return await super().on_message_send(params, context)
 
@@ -94,14 +99,18 @@ def _to_domain_message(self, message: Message) -> DomainMessage:
             Domain message with extracted text and part flags
         """
         text_parts = [
-            part.root.text
+            part.text
             for part in message.parts
-            if hasattr(part.root, 'text')
+            if part.WhichOneof("content") == "text"
         ]
         text = " ".join(text_parts) if text_parts else ""
 
-        has_file = any(isinstance(part.root, FilePart) for part in message.parts)
-        has_data = any(isinstance(part.root, DataPart) for part in message.parts)
+        has_file = any(
+            part.WhichOneof("content") in {"raw", "url"} for part in message.parts
+        )
+        has_data = any(
+            part.WhichOneof("content") == "data" for part in message.parts
+        )
 
         return DomainMessage(
             text=text,

gradient_adk/decorator.py

@@ -34,6 +34,7 @@ def _build_request_context(req: Request) -> RequestContext:
 
 from fastapi import FastAPI, HTTPException, Request
 from fastapi.responses import StreamingResponse as FastAPIStreamingResponse
+from opentelemetry.trace.propagation.tracecontext import TraceContextTextMapPropagator
 import uvicorn
 
 from .logging import get_logger
@@ -105,10 +106,7 @@ async def _submit_tracking(self, error: Optional[str]) -> None:
             return
 
         try:
-            self._tracker._req["outputs"] = "".join(self._collected)
-            if error:
-                self._tracker._req["error"] = error
-            await self._tracker._submit()
+            self._tracker.on_request_end(outputs="".join(self._collected), error=error)
         except Exception:
             # Never break streaming due to tracking errors
             pass
@@ -164,10 +162,12 @@ async def run(req: Request):
         except Exception as e:
             raise HTTPException(status_code=400, detail=f"Invalid JSON: {e}")
 
-        is_evaluation = "evaluation-id" in req.headers
+        evaluation_run_uuid = req.headers.get("evaluation-id")
+        is_evaluation = evaluation_run_uuid is not None
 
         context = _build_request_context(req)
         session_id = context.session_id
+        parent_context = TraceContextTextMapPropagator().extract(context.headers)
 
         # Initialize tracker
         tr = None
@@ -179,6 +179,8 @@ async def run(req: Request):
                     body,
                     is_evaluation=is_evaluation,
                     session_id=session_id,
+                    parent_context=parent_context,
+                    evaluation_run_uuid=evaluation_run_uuid,
                 )
         except Exception:
             pass
@@ -222,7 +224,7 @@ async def run(req: Request):
                     trace_id = None
                     if tr:
                         try:
-                            tr._req["outputs"] = result
+                            tr.on_request_end(outputs=result, error=None)
                             trace_id = await tr.submit_and_get_trace_id()
                         except Exception:
                             pass
@@ -233,9 +235,9 @@ async def run(req: Request):
                 except Exception as e:
                     if tr:
                         try:
-                            tr._req["outputs"] = "".join(collected_chunks)
-                            tr._req["error"] = str(e)
-                            await tr._submit()
+                            tr.on_request_end(
+                                outputs="".join(collected_chunks), error=str(e)
+                            )
                         except Exception:
                             pass
                     logger.error(

gradient_adk/runtime/digitalocean_tracker.py

@@ -142,6 +142,8 @@ def on_request_start(
         inputs: Dict[str, Any],
         is_evaluation: bool = False,
         session_id: Optional[str] = None,
+        parent_context: Any = None,
+        evaluation_run_uuid: Optional[str] = None,
     ) -> contextvars.Token:
         """Start tracking a new request.