diff --git a/plugins/platforms/a2a/adapter.py b/plugins/platforms/a2a/adapter.py index 39241d1b63..6b88f817e8 100644 --- a/plugins/platforms/a2a/adapter.py +++ b/plugins/platforms/a2a/adapter.py @@ -964,21 +964,25 @@ class A2AAdapter(BasePlatformAdapter): handler.wfile.write(chunk.encode("utf-8")) handler.wfile.flush() - def _emit_terminal(self, handler, task_id: str, context_id: str, state: str, reply: str) -> None: + def _emit_terminal(self, handler, task_id: str, context_id: str, state: str, reply: str, + req_id: Any = None) -> None: """Emit the final artifact/status events and close the stream (v1.0: - closure signals terminal state, no ``final`` field).""" + closure signals terminal state, no ``final`` field). + + ``req_id`` is threaded into JSON-RPC-wrapped SSE frames per §9.4.""" if reply and state == protocol.STATE_COMPLETED: self._sse_write(handler, protocol.sse_data( - protocol.artifact_update(task_id, context_id, reply))) + protocol.artifact_update(task_id, context_id, reply), req_id)) self._sse_write(handler, protocol.sse_data( - protocol.status_update(task_id, context_id, state))) + protocol.status_update(task_id, context_id, state), req_id)) else: self._sse_write(handler, protocol.sse_data( - protocol.status_update(task_id, context_id, state, reply))) + protocol.status_update(task_id, context_id, state, reply), req_id)) self._sse_write(handler, protocol.sse_done()) def _rpc_message_stream(self, handler, req_id: Any, params: dict, peer: str, agent: Optional[dict] = None) -> None: - """Handle message/stream as an SSE response of StreamResponse events.""" + """Handle message/stream as an SSE response of JSON-RPC-wrapped + StreamResponse events (A2A v1.0 §9.4).""" protocol.metrics.streams_started += 1 self._sse_headers(handler) @@ -989,19 +993,21 @@ class A2AAdapter(BasePlatformAdapter): handler, terminal["id"], terminal["contextId"], terminal["status"]["state"], protocol.extract_text(terminal.get("status", {}).get("message", {}) or {}), + req_id=req_id, ) return task_id, context_id = pending["task_id"], pending["context_id"] self._sse_write(handler, protocol.sse_data(protocol.stream_task( - protocol.build_task(task_id, context_id, protocol.STATE_SUBMITTED, created_at=pending["created_iso"])))) + protocol.build_task(task_id, context_id, protocol.STATE_SUBMITTED, created_at=pending["created_iso"])), + req_id)) self._sse_write(handler, protocol.sse_data( - protocol.status_update(task_id, context_id, protocol.STATE_WORKING))) + protocol.status_update(task_id, context_id, protocol.STATE_WORKING), req_id)) state, reply = self._await_reply( pending, keepalive=lambda: self._sse_write(handler, ": keepalive\n\n")) state, reply = self._finalize_task(pending, state, reply) - self._emit_terminal(handler, task_id, context_id, state, reply) + self._emit_terminal(handler, task_id, context_id, state, reply, req_id=req_id) except (BrokenPipeError, ConnectionResetError): logger.debug("A2A: stream client disconnected") @@ -1030,7 +1036,7 @@ class A2AAdapter(BasePlatformAdapter): state, reply = rec["state"], rec.get("reply", "") break self._sse_write(handler, ": keepalive\n\n") - self._emit_terminal(handler, task_id, rec["context_id"], state, reply) + self._emit_terminal(handler, task_id, rec["context_id"], state, reply, req_id=req_id) except (BrokenPipeError, ConnectionResetError): logger.debug("A2A: subscribe client disconnected") diff --git a/plugins/platforms/a2a/protocol.py b/plugins/platforms/a2a/protocol.py index d230ac7ad1..99fedbb241 100644 --- a/plugins/platforms/a2a/protocol.py +++ b/plugins/platforms/a2a/protocol.py @@ -372,14 +372,19 @@ def build_task( *, created_at: str = "", ) -> dict: - """Build an A2A v1.0 Task object for a message/send result.""" + """Build an A2A v1.0 Task object for a message/send result. + + ``created_at`` is accepted for call-site compatibility but not serialized — + the A2A v1.0 ``Task`` proto (``lf.a2a.v1.Task``) has no ``createdAt`` or + ``lastModified`` field. Strict ProtoJSON parsers (e.g. a2a-sdk 1.1.0) + reject unknown fields, so we must not include them. The spec's §5.6.1 + timestamp-format example mentions them but they are not in the proto. + """ now = now_iso() task: dict[str, Any] = { "id": task_id, "contextId": context_id, "status": {"state": state, "timestamp": now}, - "createdAt": created_at or now, - "lastModified": now, } if agent_text: task["status"]["message"] = text_message(ROLE_AGENT, agent_text, context_id) @@ -417,14 +422,29 @@ def artifact_update(task_id: str, context_id: str, text: str) -> dict: } -def sse_data(payload: dict) -> str: - """Encode one StreamResponse as an SSE data frame (member-name discriminated).""" - return f"data: {json.dumps(payload, ensure_ascii=False)}\n\n" +def sse_data(payload: dict, req_id: Any = None) -> str: + """Encode one StreamResponse as a JSON-RPC-wrapped SSE data frame. + + A2A v1.0 §9.4 requires each SSE frame to be a full JSON-RPC response: + ``{"jsonrpc":"2.0","id":,"result":{StreamResponse}}``. Emitting a + bare StreamResponse (the REST binding shape) breaks JSON-RPC clients that + expect the envelope, including the official a2a-sdk. + """ + if req_id is not None: + envelope = jsonrpc_result(req_id, payload) + else: + envelope = payload # legacy/fallback — no envelope + return f"data: {json.dumps(envelope, ensure_ascii=False)}\n\n" def sse_done() -> str: - """SSE stream-closure marker — terminal state is implied by closure in v1.0.""" - return "event: done\ndata: {}\n\n" + """SSE stream-closure marker — a comment, not a parseable data frame. + + A2A v1.0 signals terminal state by closing the stream. Emitting + ``data: {}`` causes JSON-RPC clients to try parsing an empty response and + fail. An SSE comment line (``: done``) is ignored by all SSE parsers. + """ + return ": done\n\n" # -------------------------------------------------------------------------- diff --git a/tests/plugins/test_a2a_phase23.py b/tests/plugins/test_a2a_phase23.py index 8bb7affa17..933ae037c9 100644 --- a/tests/plugins/test_a2a_phase23.py +++ b/tests/plugins/test_a2a_phase23.py @@ -55,7 +55,8 @@ def _make_live_adapter(monkeypatch, reply_fn=None): def _post_sse(url, body): """POST a JSON-RPC request and return the parsed SSE stream as - (data_payloads, event_names).""" + (data_payloads, event_names). Unwraps the JSON-RPC envelope from + each data frame so callers see bare StreamResponse objects.""" req = urllib.request.Request( url, data=json.dumps(body).encode(), headers={"Content-Type": "application/json"}, method="POST", @@ -66,11 +67,17 @@ def _post_sse(url, body): for block in raw.split("\n\n"): for line in block.splitlines(): if line.startswith("event: "): - events.append(line[len("event: "):].strip()) + events.append(line[len("event:"):].strip()) elif line.startswith("data: "): data = line[len("data: "):].strip() if data: - payloads.append(json.loads(data)) + obj = json.loads(data) + # Unwrap JSON-RPC envelope: {"jsonrpc":"2.0","id":...,"result":{...}} + if isinstance(obj, dict) and "jsonrpc" in obj and "result" in obj: + payloads.append(obj["result"]) + else: + payloads.append(obj) + # SSE comment lines (": done") are ignored — not data frames. return payloads, events @@ -127,8 +134,32 @@ class TestStreamResponseFormat: # No event-name line: v1.0 discriminates by member presence. assert "event:" not in chunk + def test_sse_data_jsonrpc_envelope(self): + """A2A v1.0 §9.4: SSE frames must be JSON-RPC-wrapped when req_id is + provided. Bare StreamResponse (REST binding) breaks a2a-sdk clients.""" + chunk = protocol.sse_data({"statusUpdate": {"taskId": "t"}}, req_id="42") + assert chunk.startswith("data: ") + obj = json.loads(chunk[len("data: "):].strip()) + assert obj["jsonrpc"] == "2.0" + assert obj["id"] == "42" + assert "result" in obj + assert obj["result"]["statusUpdate"]["taskId"] == "t" + + def test_sse_data_no_envelope_without_req_id(self): + """Without req_id, sse_data falls back to bare payload for legacy callers.""" + chunk = protocol.sse_data({"statusUpdate": {"taskId": "t"}}) + obj = json.loads(chunk[len("data: "):].strip()) + assert "jsonrpc" not in obj + assert obj["statusUpdate"]["taskId"] == "t" + def test_sse_done_marker(self): - assert protocol.sse_done() == "event: done\ndata: {}\n\n" + """v1.0 signals stream completion by closing the stream. The done + marker is an SSE comment (``: done``), not a parseable data frame — + emitting ``data: {}`` breaks JSON-RPC clients that try to parse it.""" + done = protocol.sse_done() + assert ": done" in done + assert "data:" not in done # no data frame for SDK to parse + assert done.endswith("\n\n") @pytest.mark.integration @@ -164,7 +195,7 @@ class TestStreamingEndToEnd: assert len(artifacts) == 1 assert "ECHO:" in protocol.extract_text(artifacts[0]["artifact"]) - assert events == ["done"] + assert events == [] # v1.0: stream closure is the terminal signal, no event frame await adapter.disconnect() asyncio.run(run()) @@ -189,7 +220,7 @@ class TestStreamingEndToEnd: artifacts = [p for p in payloads if "artifactUpdate" in p] assert artifacts and "ECHO:" in protocol.extract_text( artifacts[0]["artifactUpdate"]["artifact"]) - assert events == ["done"] + assert events == [] # v1.0: stream closure is the terminal signal, no event frame await adapter.disconnect() asyncio.run(run()) diff --git a/tests/plugins/test_a2a_plugin.py b/tests/plugins/test_a2a_plugin.py index 14db28400c..346284b277 100644 --- a/tests/plugins/test_a2a_plugin.py +++ b/tests/plugins/test_a2a_plugin.py @@ -381,8 +381,10 @@ class TestV1Task: assert task["status"]["state"] == "TASK_STATE_COMPLETED" assert task["artifacts"][0]["parts"][0] == {"text": "the answer", "mediaType": "text/plain"} assert "kind" not in task - assert task["createdAt"] - assert task["lastModified"] + # A2A v1.0 Task proto (lf.a2a.v1.Task) has no createdAt/lastModified. + # Strict ProtoJSON parsers (a2a-sdk) reject unknown fields. + assert "createdAt" not in task + assert "lastModified" not in task def test_failed_task_has_message_no_artifacts(self): task = protocol.build_task("t2", "c2", protocol.STATE_FAILED, "went wrong")