features/a2a_sse_streaming.feature

@@ -1,8 +1,8 @@
 @mock_only
 Feature: A2A SSE streaming for task updates and artifacts
   As a CleverAgents developer
-  I want the A2A event system to support SSE streaming
-  So that clients can receive real-time task status and artifact updates
+  I want the A2A event system to support SSE streaming with JSON-RPC 2.0 compliance
+  So that clients can receive real-time task status and artifact updates in a standard format
 
   Scenario: SSE event formatter produces valid text/event-stream output
     Given an A2aEvent with type "TaskStatusUpdateEvent" and plan_id "plan-001"
@@ -44,9 +44,46 @@ Feature: A2A SSE streaming for task updates and artifacts
   Scenario: TaskArtifactUpdateEvent type constant is defined
     Then the TASK_ARTIFACT_UPDATE constant should equal "TaskArtifactUpdateEvent"
 
-  Scenario: SseEventFormatter data payload is valid JSON
+  Scenario: SseEventFormatter produces JSON-RPC 2.0 compliant data payload
     Given an A2aEvent with type "TaskStatusUpdateEvent" and plan_id "plan-002"
     When I format the event as SSE
     Then the SSE data line should contain valid JSON
-    And the JSON should have key "event_type" with value "TaskStatusUpdateEvent"
-    And the JSON should have key "plan_id" with value "plan-002"
+    And the JSON should have key "jsonrpc" with value "2.0"
+    And the JSON should have key "method" with value "task/statusUpdate"
+    And the JSON should have a "params" object
+    And the JSON params should have key "taskId" with value "plan-002"
+
+  Scenario: SseEventFormatter produces JSON-RPC 2.0 for TaskArtifactUpdateEvent
+    Given an A2aEvent with type "TaskArtifactUpdateEvent" and plan_id "plan-003"
+    When I format the event as SSE
+    Then the SSE data line should contain valid JSON
+    And the JSON should have key "jsonrpc" with value "2.0"
+    And the JSON should have key "method" with value "task/artifactUpdate"
+    And the JSON should have a "params" object
+    And the JSON params should have key "taskId" with value "plan-003"
+
+  Scenario: SseEventFormatter includes event data fields in params
+    Given an A2aEvent with type "TaskStatusUpdateEvent" and plan_id "plan-004" and data {"state": "working", "progress": 0.5}
+    When I format the event as SSE
+    Then the SSE data line should contain valid JSON
+    And the JSON params should have key "state" with value "working"
+    And the JSON params should have key "progress" with numeric value 0.5
+    And the JSON params should have key "taskId" with value "plan-004"
+
+  Scenario: SseEventFormatter handles events without plan_id
+    Given an A2aEvent with type "TaskStatusUpdateEvent" and no plan_id
+    When I format the event as SSE
+    Then the SSE data line should contain valid JSON
+    And the JSON should have key "jsonrpc" with value "2.0"
+    And the JSON should have key "method" with value "task/statusUpdate"
+    And the JSON should have a "params" object
+    And the JSON params should not have key "taskId"
+
+  Scenario: SseEventFormatter excludes non-spec fields from data payload
+    Given an A2aEvent with type "TaskStatusUpdateEvent" and plan_id "plan-005"
+    When I format the event as SSE
+    Then the SSE data line should contain valid JSON
+    And the JSON should not have key "event_id"
+    And the JSON should not have key "event_type"
+    And the JSON should not have key "timestamp"
+    And the JSON should not have key "plan_id"

features/steps/a2a_sse_streaming_steps.py

@@ -7,22 +7,20 @@ from typing import Any
 from unittest.mock import MagicMock
 
 from behave import given, then, when
+from behave.runner import Context
 
-try:
-    from cleveragents.a2a.events import (
-        TASK_ARTIFACT_UPDATE,
-        TASK_STATUS_UPDATE,
-        A2aEventQueue,
-        EventBusBridge,
-        SseEventFormatter,
-    )
-    from cleveragents.a2a.models import A2aEvent
-except ImportError:
-    pass  # a2a module not available
+from cleveragents.a2a.events import (
+    TASK_ARTIFACT_UPDATE,
+    TASK_STATUS_UPDATE,
+    A2aEventQueue,
+    EventBusBridge,
+    SseEventFormatter,
+)
+from cleveragents.a2a.models import A2aEvent
 
 
 @given('an A2aEvent with type "{event_type}" and plan_id "{plan_id}"')
-def step_create_event(context: Any, event_type: str, plan_id: str) -> None:
+def step_create_event(context: Context, event_type: str, plan_id: str) -> None:
     context.event = A2aEvent(
         event_type=event_type,
         plan_id=plan_id,
@@ -30,44 +28,67 @@ def step_create_event(context: Any, event_type: str, plan_id: str) -> None:
     )
 
 
+@given(
+    'an A2aEvent with type "{event_type}" and plan_id "{plan_id}" and data {data_json}'
+)
+def step_create_event_with_data(
+    context: Context, event_type: str, plan_id: str, data_json: str
+) -> None:
+    data: dict[str, Any] = json.loads(data_json)
+    context.event = A2aEvent(
+        event_type=event_type,
+        plan_id=plan_id,
+        data=data,
+    )
+
+
+@given('an A2aEvent with type "{event_type}" and no plan_id')
+def step_create_event_no_plan_id(context: Context, event_type: str) -> None:
+    context.event = A2aEvent(
+        event_type=event_type,
+        plan_id=None,
+        data={"status": "working"},
+    )
+
+
 @when("I format the event as SSE")
-def step_format_sse(context: Any) -> None:
+def step_format_sse(context: Context) -> None:
     context.sse_output = SseEventFormatter.format(context.event)
 
 
 @then('the SSE output should contain "{text}"')
-def step_sse_contains(context: Any, text: str) -> None:
+def step_sse_contains(context: Context, text: str) -> None:
     assert text in context.sse_output, (
         f"Expected '{text}' in SSE output: {context.sse_output!r}"
     )
 
 
 @then("the SSE output should end with two newlines")
-def step_sse_ends_newlines(context: Any) -> None:
+def step_sse_ends_newlines(context: Context) -> None:
     assert context.sse_output.endswith("\n\n"), (
         f"SSE output should end with two newlines: {context.sse_output!r}"
     )
 
 
 @when("I format a keepalive SSE message")
-def step_format_keepalive(context: Any) -> None:
+def step_format_keepalive(context: Context) -> None:
     context.keepalive = SseEventFormatter.format_keepalive()
 
 
 @then('the keepalive should start with ":"')
-def step_keepalive_starts_colon(context: Any) -> None:
+def step_keepalive_starts_colon(context: Context) -> None:
     assert context.keepalive.startswith(":"), (
         f"Keepalive should start with ':', got: {context.keepalive!r}"
     )
 
 
 @given("an A2aEventQueue for SSE testing")
-def step_create_queue(context: Any) -> None:
+def step_create_queue(context: Context) -> None:
     context.queue = A2aEventQueue()
 
 
 @given("a mock EventBus")
-def step_create_mock_bus(context: Any) -> None:
+def step_create_mock_bus(context: Context) -> None:
     context.bus = MagicMock()
     context.bus_callback = None
 
@@ -79,13 +100,13 @@ def step_create_mock_bus(context: Any) -> None:
 
 
 @given("an EventBusBridge connecting bus to queue")
-def step_create_bridge(context: Any) -> None:
+def step_create_bridge(context: Context) -> None:
     context.bridge = EventBusBridge(context.bus, context.queue)
     context.bridge.start()
 
 
 @when("the bridge receives a PLAN_CREATED domain event")
-def step_bridge_plan_created(context: Any) -> None:
+def step_bridge_plan_created(context: Context) -> None:
     domain_event = MagicMock()
     domain_event.event_type = MagicMock(value="PLAN_CREATED")
     domain_event.plan_id = "plan-test-001"
@@ -98,7 +119,7 @@ def step_bridge_plan_created(context: Any) -> None:
 
 
 @when("the bridge receives a CHECKPOINT_RESTORED domain event")
-def step_bridge_checkpoint_restored(context: Any) -> None:
+def step_bridge_checkpoint_restored(context: Context) -> None:
     domain_event = MagicMock()
     domain_event.event_type = MagicMock(value="CHECKPOINT_RESTORED")
     domain_event.plan_id = "plan-test-002"
@@ -107,12 +128,12 @@ def step_bridge_checkpoint_restored(context: Any) -> None:
 
 
 @when("the queue is closed")
-def step_close_queue(context: Any) -> None:
+def step_close_queue(context: Context) -> None:
     context.queue.close()
 
 
 @then("the queue should contain a TaskStatusUpdateEvent")
-def step_queue_has_status_event(context: Any) -> None:
+def step_queue_has_status_event(context: Context) -> None:
     events = context.queue.get_events()
     types = [e.event_type for e in events]
     assert TASK_STATUS_UPDATE in types, (
@@ -121,7 +142,7 @@ def step_queue_has_status_event(context: Any) -> None:
 
 
 @then("the queue should contain a TaskArtifactUpdateEvent")
-def step_queue_has_artifact_event(context: Any) -> None:
+def step_queue_has_artifact_event(context: Context) -> None:
     events = context.queue.get_events()
     types = [e.event_type for e in events]
     assert TASK_ARTIFACT_UPDATE in types, (
@@ -130,24 +151,24 @@ def step_queue_has_artifact_event(context: Any) -> None:
 
 
 @then("no error should be raised from the bridge")
-def step_no_bridge_error(context: Any) -> None:
+def step_no_bridge_error(context: Context) -> None:
     assert context.bridge_error is None, (
         f"Expected no error, got: {context.bridge_error}"
     )
 
 
 @then('the TASK_STATUS_UPDATE constant should equal "{value}"')
-def step_check_status_constant(context: Any, value: str) -> None:
+def step_check_status_constant(context: Context, value: str) -> None:
     assert value == TASK_STATUS_UPDATE
 
 
 @then('the TASK_ARTIFACT_UPDATE constant should equal "{value}"')
-def step_check_artifact_constant(context: Any, value: str) -> None:
+def step_check_artifact_constant(context: Context, value: str) -> None:
     assert value == TASK_ARTIFACT_UPDATE
 
 
 @then("the SSE data line should contain valid JSON")
-def step_sse_data_valid_json(context: Any) -> None:
+def step_sse_data_valid_json(context: Context) -> None:
     for line in context.sse_output.splitlines():
         if line.startswith("data: "):
             payload = line[6:]
@@ -157,7 +178,50 @@ def step_sse_data_valid_json(context: Any) -> None:
 
 
 @then('the JSON should have key "{key}" with value "{value}"')
-def step_json_has_key_value(context: Any, key: str, value: str) -> None:
+def step_json_has_key_value(context: Context, key: str, value: str) -> None:
     assert context.sse_json.get(key) == value, (
         f"Expected JSON['{key}'] == '{value}', got: {context.sse_json.get(key)!r}"
     )
+
+
+@then('the JSON should have a "{key}" object')
+def step_json_has_object(context: Context, key: str) -> None:
+    assert key in context.sse_json, (
+        f"Expected JSON to have key '{key}', got keys: {list(context.sse_json.keys())}"
+    )
+    assert isinstance(context.sse_json[key], dict), (
+        f"Expected JSON['{key}'] to be a dict, got: {type(context.sse_json[key])!r}"
+    )
+    context.sse_params = context.sse_json[key]
+
+
+@then('the JSON params should have key "{key}" with value "{value}"')
+def step_json_params_has_key_value(context: Context, key: str, value: str) -> None:
+    params: dict[str, Any] = context.sse_json.get("params", {})
+    assert params.get(key) == value, (
+        f"Expected params['{key}'] == '{value}', got: {params.get(key)!r}"
+    )
+
+
+@then('the JSON params should have key "{key}" with numeric value {value:g}')
+def step_json_params_has_numeric_value(
+    context: Context, key: str, value: float
+) -> None:
+    params: dict[str, Any] = context.sse_json.get("params", {})
+    actual = params.get(key)
+    assert actual == value, f"Expected params['{key}'] == {value}, got: {actual!r}"
+
+
+@then('the JSON params should not have key "{key}"')
+def step_json_params_not_have_key(context: Context, key: str) -> None:
+    params: dict[str, Any] = context.sse_json.get("params", {})
+    assert key not in params, (
+        f"Expected params to NOT have key '{key}', but found: {params.get(key)!r}"
+    )
+
+
+@then('the JSON should not have key "{key}"')
+def step_json_not_have_key(context: Context, key: str) -> None:
+    assert key not in context.sse_json, (
+        f"Expected JSON to NOT have key '{key}', but found: {context.sse_json.get(key)!r}"
+    )

src/cleveragents/a2a/events.py

@@ -6,6 +6,8 @@ local mode and a stub for remote subscriptions that raises
 
 :class:`SseEventFormatter` converts :class:`A2aEvent` instances into
 ``text/event-stream`` formatted strings per the Server-Sent Events spec.
+The data payload follows JSON-RPC 2.0 notification format as required by
+the A2A protocol specification.
 
 :class:`EventBusBridge` subscribes to the internal ``EventBus`` and
 publishes translated :class:`A2aEvent` instances to an event queue.
@@ -15,7 +17,7 @@ from __future__ import annotations
 
 import json
 from collections.abc import Callable
-from typing import Any
+from typing import Any, ClassVar
 
 import structlog
 from ulid import ULID
@@ -146,18 +148,55 @@ class SseEventFormatter:
         data: <json payload>
 
     Two trailing newlines terminate the event per the EventSource spec.
+    The data payload follows JSON-RPC 2.0 notification format per the A2A
+    protocol specification (§Streaming Architecture).
     """
 
+    # Mapping from A2A event type names to JSON-RPC 2.0 method strings.
+    # Per spec §Streaming Architecture: TaskStatusUpdateEvent → task/statusUpdate,
+    # TaskArtifactUpdateEvent → task/artifactUpdate.
+    _EVENT_TYPE_TO_METHOD: ClassVar[dict[str, str]] = {
+        "TaskStatusUpdateEvent": "task/statusUpdate",
+        "TaskArtifactUpdateEvent": "task/artifactUpdate",
+    }
+
     @staticmethod
     def format(event: A2aEvent) -> str:
-        """Format an :class:`A2aEvent` as an SSE text block."""
+        """Format an :class:`A2aEvent` as an SSE text block.
+
+        The ``data:`` line contains a JSON-RPC 2.0 notification envelope::
+
+            {
+                "jsonrpc": "2.0",
+                "method": "task/statusUpdate",
+                "params": {
+                    "taskId": "<plan_id>",
+                    ...event data fields...
+                }
+            }
+
+        Non-spec fields (``event_id``, ``event_type``, ``timestamp``) are
+        excluded from the data payload; they are carried in the SSE envelope
+        headers (``event:`` and ``id:`` lines) instead.
+        """
+        # Resolve JSON-RPC 2.0 method name from event type.
+        method = SseEventFormatter._EVENT_TYPE_TO_METHOD.get(
+            event.event_type,
+            f"task/{event.event_type}",
+        )
+
+        # Build params from event data fields.
+        params: dict[str, Any] = dict(event.data)
+
+        # Include taskId in params when plan_id is present (per spec).
+        if event.plan_id:
+            params["taskId"] = event.plan_id
+
         data_payload = json.dumps(
             {
-                "event_id": event.event_id,
-                "event_type": event.event_type,
-                "plan_id": event.plan_id,
-                "data": event.data,
-                "timestamp": event.timestamp,
+                "jsonrpc": "2.0",
+                "method": method,
+                "params": params,
             },
             default=str,
         )