fix(a2a): reformat SseEventFormatter output to JSON-RPC 2.0 notification structure

- Updated SseEventFormatter.format() to produce JSON-RPC 2.0 compliant data payload
- Added mapping from A2A event types to JSON-RPC method names (task/statusUpdate, task/artifactUpdate)
- Moved event data fields into params object, with taskId included when plan_id is present
- Updated BDD tests to verify JSON-RPC 2.0 structure compliance
- Added test scenarios for events with/without plan_id and with custom data

Fixes #1502

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,37 @@ 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 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 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"

features/steps/a2a_sse_streaming_steps.py

@@ -161,3 +161,59 @@ def step_json_has_key_value(context: Any, key: str, value: str) -> None:
     assert context.sse_json.get(key) == value, (
         f"Expected JSON['{key}'] == '{value}', got: {context.sse_json.get(key)!r}"
     )
+
+
+@given('an A2aEvent with type "{event_type}" and plan_id "{plan_id}" and data {data_json}')
+def step_create_event_with_data(context: Any, event_type: str, plan_id: str, data_json: str) -> None:
+    data = 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(context: Any, event_type: str) -> None:
+    context.event = A2aEvent(
+        event_type=event_type,
+        plan_id=None,
+        data={"status": "working"},
+    )
+
+
+@then('the JSON should have a "{key}" object')
+def step_json_has_object(context: Any, key: str) -> None:
+    assert key in context.sse_json, (
+        f"Expected JSON to have key '{key}', got: {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])}"
+    )
+
+
+@then('the JSON params should have key "{key}" with value "{value}"')
+def step_json_params_has_string(context: Any, key: str, value: str) -> None:
+    params = 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 value {value}')
+def step_json_params_has_value(context: Any, key: str, value: str) -> None:
+    params = context.sse_json.get("params", {})
+    # Parse numeric values
+    expected = json.loads(value)
+    actual = params.get(key)
+    assert actual == expected, (
+        f"Expected params['{key}'] == {expected}, got: {actual!r}"
+    )
+
+
+@then('the JSON params should not have key "{key}"')
+def step_json_params_missing_key(context: Any, key: str) -> None:
+    params = context.sse_json.get("params", {})
+    assert key not in params, (
+        f"Expected params not to have key '{key}', but it does: {params}"
+    )

src/cleveragents/a2a/events.py

@@ -6,6 +6,7 @@ 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.
 
 :class:`EventBusBridge` subscribes to the internal ``EventBus`` and
 publishes translated :class:`A2aEvent` instances to an event queue.
@@ -146,21 +147,45 @@ 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.
     """
 
+    # Mapping from A2A event types to JSON-RPC 2.0 method names
+    _EVENT_TYPE_TO_METHOD = {
+        "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 payload is formatted as a JSON-RPC 2.0 notification:
+        {"jsonrpc": "2.0", "method": "...", "params": {...}}
+        """
+        # Map event_type to JSON-RPC method name
+        method = SseEventFormatter._EVENT_TYPE_TO_METHOD.get(
+            event.event_type,
+            f"task/{event.event_type}",  # fallback for unknown types
+        )
+        
+        # Build params object from event fields
+        params: dict[str, Any] = dict(event.data)  # Start with event.data contents
+        
+        # Add taskId if plan_id is present
+        if event.plan_id:
+            params["taskId"] = event.plan_id
+        
+        # Build JSON-RPC 2.0 notification envelope
         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,
         )
+        
         lines = [
             f"event: {event.event_type}",
             f"id: {event.event_id}",