diff --git a/docs/development/custom-ui/a2a-client.mdx b/docs/development/custom-ui/a2a-client.mdx index c9be9e8c..b2caa713 100644 --- a/docs/development/custom-ui/a2a-client.mdx +++ b/docs/development/custom-ui/a2a-client.mdx @@ -107,9 +107,11 @@ const stream = client.sendMessageStream({ ### 3. Handle streaming updates and show the form -Use `handleTaskStatusUpdate` to detect form requests and capture the task ID so you can continue the same task. Streamed output arrives primarily via `statusUpdate` events; `message` events are the fallback for non-streaming agents. +Use `handleTaskStatusUpdate` to detect form requests and capture the task ID so you can continue the same task. The stream yields `StreamResponse` objects — a union of `{ statusUpdate }`, `{ message }`, `{ artifactUpdate }`, and `{ task }`. Use `"statusUpdate" in event` checks to discriminate. -The stream yields `StreamResponse` objects — a union of `{ statusUpdate }`, `{ message }`, `{ artifactUpdate }`, and `{ task }`. Use `"statusUpdate" in event` checks to discriminate. +When the streaming extension is active, incremental output arrives as JSON Patch operations in `statusUpdate.metadata`. Extract and apply these patches to a local draft to reconstruct the message progressively, then update your UI with the draft's parts. When no patches are present (final status update or non-streaming agents), fall back to `statusUpdate.status.message`. A standalone `message` event may also arrive at task completion. + +See **[Streaming](../sdk/streaming)** for the full wire format and patch operations. **Streaming text must live in persistent state, not a temporary variable** @@ -119,8 +121,10 @@ The stream yields `StreamResponse` objects — a union of `{ statusUpdate }`, `{ ```typescript import { handleTaskStatusUpdate, type TaskStatusUpdateType } from "@kagenti/adk"; +import { extractStreamingPatches, applyPatches } from "@kagenti/adk"; let taskId: string | undefined; +const draft: Record = {}; // 1. Add a placeholder agent message to the message list before streaming const placeholderId = crypto.randomUUID(); @@ -141,7 +145,15 @@ for await (const event of stream) { } } - // 2. Update the placeholder in-place as streaming text arrives + // 2. Apply streaming patches to reconstruct the message incrementally + const patches = extractStreamingPatches(event.statusUpdate.metadata); + if (patches) { + applyPatches(draft, patches); + updateMessage(placeholderId, { parts: draft.parts ?? [] }); + continue; + } + + // 3. No patches — final status update or non-streaming agent; use status message const msg = event.statusUpdate.status?.message; if (msg) { updateMessage(placeholderId, { @@ -151,7 +163,7 @@ for await (const event of stream) { } } - // 3. If a final message arrives, replace placeholder content (non-streaming agents) + // 4. Standalone message (non-streaming agents or final response) if ("message" in event && event.message) { updateMessage(placeholderId, { parts: event.message.parts, @@ -160,7 +172,7 @@ for await (const event of stream) { } } -// 4. Mark streaming complete +// 5. Mark streaming complete updateMessage(placeholderId, { isStreaming: false }); ``` @@ -168,8 +180,8 @@ updateMessage(placeholderId, { isStreaming: false }); **Why task status updates for chat streaming** 1. A2A uses tasks to represent long running work so you can track progress and cancel. - 2. Status updates provide the incremental channel for UI events and streaming output. - 3. Each streamed token typically arrives as a `TaskStatusUpdateEvent` in Kagenti ADK, so you can append text as it arrives. + 2. Status updates carry incremental JSON Patch operations in their metadata, which the client applies to a local draft to reconstruct the message token by token. + 3. Clients that don't support the streaming extension simply ignore the metadata and render the final complete message at task completion. @@ -225,15 +237,27 @@ Stream the response using the same placeholder pattern from step 3. Handle artif ```typescript const placeholderId = crypto.randomUUID(); +const draft: Record = {}; addMessage({ id: placeholderId, role: "agent", parts: [], isStreaming: true }); for await (const event of responseStream) { - if ("statusUpdate" in event && event.statusUpdate?.status?.message) { - const msg = event.statusUpdate.status.message; - updateMessage(placeholderId, { - parts: msg.parts, - metadata: msg.metadata, - }); + if ("statusUpdate" in event && event.statusUpdate) { + // Apply streaming patches when present + const patches = extractStreamingPatches(event.statusUpdate.metadata); + if (patches) { + applyPatches(draft, patches); + updateMessage(placeholderId, { parts: draft.parts ?? [] }); + continue; + } + + // Final status update — use status message + const msg = event.statusUpdate.status?.message; + if (msg) { + updateMessage(placeholderId, { + parts: msg.parts, + metadata: msg.metadata, + }); + } } if ("artifactUpdate" in event && event.artifactUpdate) { @@ -277,13 +301,13 @@ for await (const event of responseStream) { Kagenti ADK’s A2A streaming is task-based. A single `sendMessageStream` call yields `StreamResponse` objects — a union with these shapes: - `{ task }`: the initial Task object. Capture `task.id`, and use `task.status` (and optional `history`/`artifacts`) to seed your UI state. -- `{ statusUpdate }`: a task status transition. When the agent is streaming, incremental output is typically delivered via `event.statusUpdate.status.message`. +- `{ statusUpdate }`: a task status transition. When the streaming extension is active, incremental output is delivered as JSON Patch operations in `event.statusUpdate.metadata`. Extract and apply these patches to a local draft to reconstruct the message progressively. When no patches are present, `event.statusUpdate.status.message` contains the complete status message. See **[Streaming](../sdk/streaming)** for the full wire format. - `{ artifactUpdate }`: artifacts as they are generated or updated (useful for streamed files, canvases, or structured outputs). -- `{ message }`: a standalone Message. This is common for non‑streaming agents and may also appear as a final response. +- `{ message }`: a standalone Message at task completion. Non‑streaming agents also use this event. Use `"statusUpdate" in event` to discriminate between shapes. -Key detail for Kagenti ADK streaming: **incremental output usually arrives inside `statusUpdate` events** (`event.statusUpdate.status.message`). If you only render `message` events, you may miss streamed output. +Key detail for Kagenti ADK streaming: **incremental output arrives as JSON Patch operations inside `statusUpdate` event metadata**, not as full messages. Use `extractStreamingPatches` and `applyPatches` to apply them to a local draft. If you only render `message` events, you will miss streamed output. ## Handling failed states @@ -321,5 +345,6 @@ For more about error handling, see **[Error Handling](./error-handling)**. - **Using `@a2a-js/sdk` `ClientFactory`**: the Kagenti ADK platform proxy serves agent cards without the `url` field that `@a2a-js/sdk` requires. Use `buildAgentClient` from `@kagenti/adk` instead. - **Wrong token in A2A requests**: use the *context token* for A2A fetches, not the user access token. - **Missing metadata merge**: merge agent card fulfillments with user metadata when you send responses. -- **Streaming text vanishes**: many agents never emit a `message` event — the entire response arrives via `statusUpdate`. Use the placeholder pattern (see step 3) so accumulated text lives in the message list, not ephemeral state. +- **Streaming text vanishes**: many agents never emit a `message` event — the entire response arrives as JSON Patch operations in `statusUpdate` metadata. Use the placeholder pattern (see step 3) so accumulated text lives in the message list, not ephemeral state. Extract and apply patches with `extractStreamingPatches` / `applyPatches`. +- **Ignoring status updates**: streamed agent output arrives as patches inside `statusUpdate` event metadata, not as `message` events. Only render `message` for the final complete response or non-streaming agents. - **Node fetch missing**: Node < 18 requires a `fetch` polyfill or custom `fetch` passed to the API client. diff --git a/docs/development/sdk/streaming.mdx b/docs/development/sdk/streaming.mdx new file mode 100644 index 00000000..4135a931 --- /dev/null +++ b/docs/development/sdk/streaming.mdx @@ -0,0 +1,333 @@ +--- +title: "Streaming" +description: "Understand the delta-based streaming protocol that delivers tokens to clients in real time using JSON Patch." +--- + +Kagenti ADK uses a delta-based streaming protocol to deliver agent output to clients token by token. When your agent yields strings, parts, or metadata, the SDK automatically converts them into incremental [JSON Patch (RFC 6902)](https://datatracker.ietf.org/doc/html/rfc6902) operations extended with a custom `str_ins` operation for efficient text insertion. Clients apply these patches to reconstruct the message progressively. + +This is fully transparent to agent code — you yield values as described in [Messages](./messages), and the SDK handles streaming automatically. This page explains the underlying protocol for custom UI developers and advanced use cases. + +## How it works + + + + + Your agent yields strings, `Part` objects, `Metadata`, or convenience wrappers like `AgentMessage`. + + + + + The `MessageAccumulator` collects yielded values into a draft message and emits JSON Patch operations for each change. + + + + + Each patch is sent as a `TaskStatusUpdateEvent` with the patch list in its `metadata` under the streaming extension URI. + + + + + The client applies patches to a local draft, reconstructing the message incrementally. When the task completes, the final full message is also sent for reconciliation. + + + + +## Extension negotiation + +Streaming is an opt-in A2A extension. Clients that support it include the extension in their request; agents that support it advertise it in their agent card capabilities. + +| Side | Mechanism | +| :--- | :--- | +| **Server** | Advertises `streaming: true` in the agent card capabilities. The SDK does this automatically. | +| **Client** | Includes the streaming extension URI in the request. The `@kagenti/adk` TypeScript helpers do this automatically. | + +If the client does not request the extension, the server sends only the final complete message at task completion — fully backward compatible. Consumer code using the `StreamingExtensionClient` (Python) or the `@kagenti/adk` helpers (TypeScript) works identically in both cases. + +**Extension URI:** +``` +https://a2a-extensions.adk.kagenti.dev/ui/streaming/v1 +``` + +## Wire format + +Streaming patches are carried in `TaskStatusUpdateEvent.metadata` under the extension URI key. Each update contains a list of JSON Patch operations and a `message_id` for client-side correlation. + +```json +{ + "status": { "state": "working" }, + "metadata": { + "https://a2a-extensions.adk.kagenti.dev/ui/streaming/v1": { + "message_update": [ + { "op": "replace", "path": "", "value": { "message_id": "abc-123", "parts": [{ "text": "Hello " }] } } + ], + "message_id": "abc-123" + } + } +} +``` + +Subsequent token updates use `str_ins`: + +```json +{ + "message_update": [ + { "op": "str_ins", "path": "/parts/0/text", "pos": 6, "value": "world" } + ], + "message_id": "abc-123" +} +``` + +### Patch operations + +The protocol uses three JSON Patch operations: + +| Operation | When used | Example | +| :--- | :--- | :--- | +| `replace` | Initialize the message draft (root replace) or update a field | `{ "op": "replace", "path": "", "value": { "message_id": "...", "parts": [...] } }` | +| `add` | Append a new part to the message or add metadata | `{ "op": "add", "path": "/parts/-", "value": { "text": "new part" } }` | +| `str_ins` | Insert text at a position in an existing string (token streaming) | `{ "op": "str_ins", "path": "/parts/0/text", "pos": 6, "value": "world" }` | + + + +The `str_ins` operation is a custom extension inspired by [json-crdt-patch](https://jsonjoy.com/specs/json-crdt-patch). Standard RFC 6902 operations (`remove`, `move`, `copy`, `test`) are not used by the protocol. Clients that cannot handle `str_ins` can fall back to rendering the final complete message. + + + +### `str_ins` operation + +The `str_ins` operation inserts text into an existing string value at a specific position: + +```json +{ "op": "str_ins", "path": "/parts/0/text", "pos": 5, "value": "inserted text" } +``` + +| Field | Type | Description | +| :--- | :--- | :--- | +| `op` | `"str_ins"` | Operation identifier | +| `path` | string | JSON Pointer to the target string | +| `pos` | number | Character index to insert at. If omitted, appends to end | +| `value` | string | Text to insert | + +The insertion logic is: `result = str[:pos] + value + str[pos:]` + +### Patch type definitions + + + + +```python +from kagenti_adk.types import JsonPatchOp, JsonPatch + +# JsonPatchOp is a TypedDict: +# { +# "op": str, # "replace" | "add" | "str_ins" (required) +# "path": str, # JSON Pointer path (required) +# "value": JsonValue # Patch value (optional) +# "pos": int # str_ins insertion position (optional) +# } +# +# JsonPatch = list[JsonPatchOp] +``` + + + + +```typescript +import type { StreamingPatch } from "@kagenti/adk"; + +// StreamingPatch (Zod-validated): +// { +// op: string; +// path: string; +// value?: unknown; +// pos?: number; +// } +``` + + + + +## Server-side: MessageAccumulator + +The `MessageAccumulator` is a 3-level state machine that converts agent yields into streaming patches. It runs automatically inside the SDK — agent code does not interact with it directly. + +### State machine + +``` +┌─────────────────────────────────────────────────────────┐ +│ Base Level (idle) │ +│ • Passthrough: Message, TaskStatus → no accumulation │ +│ • str / Part / Metadata / dict → enter MessageContext │ +└─────────────────────┬───────────────────────────────────┘ + │ accumulating yield + ▼ +┌─────────────────────────────────────────────────────────┐ +│ MessageContext │ +│ • Part → add to parts, emit add/replace patch │ +│ • Metadata → merge metadata, emit incremental patch │ +│ • dict → convert to DataPart, emit add patch │ +│ • str → enter TextPartContext │ +│ • Control yield → flush draft, return to Base │ +└─────────────────────┬───────────────────────────────────┘ + │ string yield + ▼ +┌─────────────────────────────────────────────────────────┐ +│ TextPartContext │ +│ • str → append chunk, emit str_ins patch │ +│ • Part / Metadata → finalize text part, return to │ +│ MessageContext │ +│ • Control yield → flush all, return to Base │ +└─────────────────────────────────────────────────────────┘ +``` + +### Example: what the accumulator emits + +When an agent yields three string chunks: + +```python +yield "Hello " +yield "world" +yield "!" +``` + +The accumulator emits these patches: + +```json +// First chunk: root replace (initializes the draft) +{ "op": "replace", "path": "", "value": { "message_id": "abc-123", "parts": [{ "text": "Hello " }] } } + +// Second chunk: str_ins (appends to existing text) +{ "op": "str_ins", "path": "/parts/0/text", "pos": 6, "value": "world" } + +// Third chunk: str_ins +{ "op": "str_ins", "path": "/parts/0/text", "pos": 11, "value": "!" } +``` + +### Mixed yields + +You can freely mix strings, parts, and metadata in a single agent turn. The accumulator handles transitions between contexts: + +```python +yield "Thinking..." # starts TextPartContext, emits replace +yield Part(text="Here is ") # finalizes text part, adds new Part +yield "the answer." # starts new TextPartContext for next text +yield Metadata({"source": "web"}) # finalizes text, emits metadata patch +``` + +## Client-side consumption + +### Python client + +The `StreamingExtensionClient` wraps raw A2A event streams into a unified delta-based API. It works identically whether the server supports streaming or not. + +```python +from kagenti_adk.a2a.extensions.streaming import ( + StreamingExtensionClient, + StreamingExtensionSpec, + TextDelta, + PartDelta, + MetadataDelta, + ArtifactDelta, + StateChange, +) +from a2a.types import TaskState + + +async def consume_stream(a2a_client, msg): + spec = StreamingExtensionSpec() + streaming = StreamingExtensionClient(spec) + + async for delta, task in streaming.stream(a2a_client.send_message(msg)): + match delta: + case TextDelta(part_index=idx, delta=text): + # Append text to part at idx + print(text, end="", flush=True) + case PartDelta(part_index=idx, part=part): + # New part added at idx + print(part) + case MetadataDelta(metadata=meta): + # Metadata updated + print(meta) + case ArtifactDelta(event=evt): + # Artifact update + print(evt) + case StateChange(state=state, message=msg): + if state == TaskState.TASK_STATE_COMPLETED: + print() # done +``` + +**Delta types:** + +| Type | Fields | Description | +| :--- | :--- | :--- | +| `TextDelta` | `part_index`, `delta` | A text chunk inserted into an existing part | +| `PartDelta` | `part_index`, `part` | A new part added to the message | +| `MetadataDelta` | `metadata` | Message metadata added or updated | +| `ArtifactDelta` | `event` | An artifact update event | +| `StateChange` | `state`, `message` | A task state transition (working, completed, etc.) | + +### TypeScript client + +The `@kagenti/adk` package provides helpers for extracting and applying streaming patches in the browser. + +```typescript +import { extractStreamingPatches, applyPatches } from "@kagenti/adk"; + +let draft: Record = {}; + +for await (const event of stream) { + if (event.kind === "status-update") { + const patches = extractStreamingPatches(event.status.message?.metadata); + + if (patches) { + draft = applyPatches(draft, patches); + renderDraft(draft); + } + } +} +``` + +`applyPatches` mutates the draft in place and returns it. It supports `replace`, `add`, and `str_ins` operations. + + + +The Kagenti ADK UI (`adk-ui`) already handles streaming patch application automatically. These helpers are for custom UI implementations. + + + +## Artifact streaming + +Artifacts use a separate chunked streaming mechanism via `ArtifactChunk`. Unlike message streaming, artifacts bypass the `MessageAccumulator` and are handled directly. + +```python +from a2a.types import Part +from kagenti_adk.a2a.types import ArtifactChunk + + +# @server.agent(...) +async def my_agent(): + artifact_id = "doc-001" + + # Stream artifact in chunks + yield ArtifactChunk( + parts=[Part(text="Chapter 1: Introduction\n")], + artifact_id=artifact_id, + name="report.md", + ) + + yield ArtifactChunk( + parts=[Part(text="Chapter 2: Methods\n")], + artifact_id=artifact_id, + last_chunk=True, # signals the final chunk + ) +``` + +Set `last_chunk=True` on the final chunk to signal completion. The server appends chunks with the same `artifact_id` until it sees the last chunk. + +## Backward compatibility + +The streaming protocol is fully backward compatible: + +- **Non-streaming clients** receive the complete final message at `TASK_STATE_COMPLETED`, as before. Streaming patches in metadata are ignored. +- **Non-streaming servers** send full messages. The `StreamingExtensionClient` decomposes these into the same `PartDelta` / `MetadataDelta` / `StateChange` deltas, so consumer code is identical. +- **Reconciliation**: When a full message arrives whose `message_id` was already streamed via patches, the client suppresses duplicate parts and only emits any new parts beyond the streamed prefix. diff --git a/docs/docs.json b/docs/docs.json index 78698e3b..0ece921a 100644 --- a/docs/docs.json +++ b/docs/docs.json @@ -126,6 +126,7 @@ "development/sdk/overview", "development/sdk/building-agents", "development/sdk/messages", + "development/sdk/streaming", "development/sdk/multi-turn", "development/sdk/files", "development/sdk/agent-details",