diff --git a/api/apps/sdk/session.py b/api/apps/sdk/session.py index 7938841838..190314fa57 100644 --- a/api/apps/sdk/session.py +++ b/api/apps/sdk/session.py @@ -597,11 +597,12 @@ async def agent_completions(tenant_id, agent_id): reference.update(ans["data"]["reference"]) if ans.get("event") == "node_finished": - node_out = ans.get("data", {}).get("outputs", {}) - if node_out.get("structured"): - structured_output = node_out["structured"] + data = ans.get("data", {}) + node_out = data.get("outputs", {}) + component_id = data.get("component_id") + if component_id is not None and "structured" in node_out: + structured_output[component_id] = copy.deepcopy(node_out["structured"]) if return_trace: - data = ans.get("data", {}) trace_items.append( { "component_id": data.get("component_id"), diff --git a/docs/references/http_api_reference.md b/docs/references/http_api_reference.md index cfefe5afa9..2a9fc40106 100644 --- a/docs/references/http_api_reference.md +++ b/docs/references/http_api_reference.md @@ -3914,16 +3914,16 @@ Asks a specified agent a question to start an AI-powered conversation. - `"session_id"`: `string` (optional) - `"inputs"`: `object` (optional) - `"user_id"`: `string` (optional) - - `"return_trace"`: `boolean` (optional, default `false`) — include execution trace logs. + - `"return_trace"`: `boolean` (optional, default `false`) — whether to include execution trace logs. See the `node_finished` event. - `"release"`: `boolean` (optional, default `false`) - whether to visit the latest published canvas. #### Streaming events to handle -When `stream=true`, the server sends Server-Sent Events (SSE). Clients should handle these `event` types: +When `stream=true`, the server sends Server-Sent Events (SSE). A client should handle these events: -- `message`: streaming content from Message components. -- `message_end`: end of a Message component; may include `reference`/`attachment`. -- `node_finished`: a component finishes; `data.inputs/outputs/error/elapsed_time` describe the node result. If `return_trace=true`, the trace is attached inside the same `node_finished` event (`data.trace`). +- `message`: Streaming content from the **Message** components. +- `message_end`: End of a **Message** component, which may include `reference`/`attachment`. +- `node_finished`: A component finishes; `data.inputs/outputs/error/elapsed_time` describes the node result. If a component produces structured output, read it from that component's `data.outputs.structured`. If `return_trace=true`, the trace is attached inside the same `node_finished` event (`data.trace`). The stream terminates with `[DONE]`. @@ -4203,6 +4203,8 @@ When `extra_body.reference_metadata.include` is `true`, each reference chunk may Non-stream: +If one or more components produce structured output, ensure you set `return_trace=true` and check each component's structured output via `trace`. The top-level `data.structured` field is a shortcut aggregated by `component_id`. + ```json { "code": 0, diff --git a/test/testcases/test_http_api/test_session_management/test_session_sdk_routes_unit.py b/test/testcases/test_http_api/test_session_management/test_session_sdk_routes_unit.py index 109889b4b6..c305cc20db 100644 --- a/test/testcases/test_http_api/test_session_management/test_session_sdk_routes_unit.py +++ b/test/testcases/test_http_api/test_session_management/test_session_sdk_routes_unit.py @@ -862,7 +862,18 @@ def test_agent_completions_stream_and_nonstream_unit(monkeypatch): async def _agent_stream(*_args, **_kwargs): yield "data:not-json" - yield "data:" + json.dumps({"event": "node_finished", "data": {"component_id": "c1"}}) + yield "data:" + json.dumps( + { + "event": "node_finished", + "data": {"component_id": "c1", "outputs": {"structured": {"alpha": 1}}}, + } + ) + yield "data:" + json.dumps( + { + "event": "node_finished", + "data": {"component_id": "c2", "outputs": {"structured": {}}}, + } + ) yield "data:" + json.dumps({"event": "other", "data": {}}) yield "data:" + json.dumps({"event": "message", "data": {"content": "hello"}}) @@ -878,14 +889,36 @@ def test_agent_completions_stream_and_nonstream_unit(monkeypatch): async def _agent_nonstream(*_args, **_kwargs): yield "data:" + json.dumps({"event": "message", "data": {"content": "A", "reference": {"doc": "r"}}}) - yield "data:" + json.dumps({"event": "node_finished", "data": {"component_id": "c2"}}) + yield "data:" + json.dumps( + { + "event": "node_finished", + "data": {"component_id": "c2", "outputs": {"structured": {"foo": "bar"}}}, + } + ) + yield "data:" + json.dumps( + { + "event": "node_finished", + "data": {"component_id": "c3", "outputs": {"structured": {"baz": 1}}}, + } + ) + yield "data:" + json.dumps( + { + "event": "node_finished", + "data": {"component_id": "c4", "outputs": {"structured": {}}}, + } + ) monkeypatch.setattr(module, "agent_completion", _agent_nonstream) monkeypatch.setattr(module, "get_request_json", lambda: _AwaitableValue({"stream": False, "return_trace": True})) res = _run(inspect.unwrap(module.agent_completions)("tenant-1", "agent-1")) assert res["data"]["data"]["content"] == "A" assert res["data"]["data"]["reference"] == {"doc": "r"} - assert res["data"]["data"]["trace"][0]["component_id"] == "c2" + assert res["data"]["data"]["structured"] == { + "c2": {"foo": "bar"}, + "c3": {"baz": 1}, + "c4": {}, + } + assert [item["component_id"] for item in res["data"]["data"]["trace"]] == ["c2", "c3", "c4"] async def _agent_nonstream_broken(*_args, **_kwargs): yield "data:{"