diff --git a/features/a2a_sse_streaming.feature b/features/a2a_sse_streaming.feature index 7c6aab42f..9c5c7313e 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,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" diff --git a/features/steps/a2a_sse_streaming_steps.py b/features/steps/a2a_sse_streaming_steps.py index ecf0a5d93..0baddc434 100644 --- a/features/steps/a2a_sse_streaming_steps.py +++ b/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}" + ) diff --git a/src/cleveragents/a2a/events.py b/src/cleveragents/a2a/events.py index 1fdbdfebc..8698dfe6e 100644 --- a/src/cleveragents/a2a/events.py +++ b/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: 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}",