From 288ff276b35e68f67f80db717b7de3b63105702b Mon Sep 17 00:00:00 2001 From: Jeffrey Phillips Freeman Date: Thu, 2 Apr 2026 23:53:13 +0000 Subject: [PATCH] fix(a2a): reformat SseEventFormatter output to JSON-RPC 2.0 notification structure MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Add _EVENT_TYPE_TO_METHOD class-level mapping (ClassVar[dict[str, str]]) to convert A2A event types to JSON-RPC 2.0 method names: TaskStatusUpdateEvent → task/statusUpdate TaskArtifactUpdateEvent → task/artifactUpdate - Refactor SseEventFormatter.format() to produce JSON-RPC 2.0 notification envelope: {"jsonrpc": "2.0", "method": "...", "params": {...}} - Move event data fields into params object; include taskId (from plan_id) in params when plan_id is present, per spec §Streaming Architecture - Remove non-spec fields (event_id, event_type, timestamp, plan_id) from the data payload; these remain in SSE envelope headers (event: and id:) - Update BDD feature to verify JSON-RPC 2.0 structure for both event types, events with/without plan_id, custom data in params, and exclusion of non-spec fields - Fix pre-existing type errors in step definitions: replace try/except ImportError pattern with direct imports and use behave.runner.Context for proper static typing (0 pyright errors) ISSUES CLOSED: #1502 --- features/a2a_sse_streaming.feature | 47 +++++++- features/steps/a2a_sse_streaming_steps.py | 124 ++++++++++++++++------ src/cleveragents/a2a/events.py | 53 +++++++-- 3 files changed, 182 insertions(+), 42 deletions(-) diff --git a/features/a2a_sse_streaming.feature b/features/a2a_sse_streaming.feature index 7c6aab42f..80bd2554a 100644 --- a/features/a2a_sse_streaming.feature +++ b/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" diff --git a/features/steps/a2a_sse_streaming_steps.py b/features/steps/a2a_sse_streaming_steps.py index ecf0a5d93..8b444b542 100644 --- a/features/steps/a2a_sse_streaming_steps.py +++ b/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}" + ) diff --git a/src/cleveragents/a2a/events.py b/src/cleveragents/a2a/events.py index 1fdbdfebc..f17ae21e7 100644 --- a/src/cleveragents/a2a/events.py +++ b/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: 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": "", + ...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, ) -- 2.52.0