Skip to content

Commit a0b0ee0

Browse files
l0lawrenceCopilot
andauthored
feat(http-client-python): generate structured JSONL/SSE streaming (#11594)
- Generate Azure-flavor client methods that return `Stream[T]` / `AsyncStream[T]` for JSONL (`application/jsonl`) and SSE (`text/event-stream`) response streams ```python stream = client.receive() # Stream[Thing] for thing in stream: ... ``` Supports this [spec](https://gist.github.com/chrisradek/b2656d7ee5db1b29768d50db0628f033#typespecevents) Missing support for : #11761 QUESTIONS/NOTES FOR NOW OR FUTURE REF: - handling multiple unamed events correctly -- supporting [@Discriminator to figure out which event is which](https://gist.github.com/chrisradek/b2656d7ee5db1b29768d50db0628f033#typespecevents) - handling `@data` (w/ w/o envelopes correctly -- python is just returning data aka Stream[T] not Stream[SSEEvent[T]] (like paging) so Stream[`@data` payload]) - decoding payload content-type correctly -- use of isEventEnvelope*? -- don't currently use for python b/c of above design - Currently not supporting automatic reconnect (last-event-id can be passed in by user if needed, but `@retry` not used - both should be exposed on Stream) -- #11761 - terminal events yielding -- [DONE] vs a terminal event that is a model and how we handle that /yield them or not EXAMPLES: What search retrieveStream() looks like: <img width="2795" height="1445" alt="image" src="https://github.com/user-attachments/assets/fe39b2fb-1654-4ec4-871e-2d8683e3fa65" /> `Stream(response=response, deserialization_callback=_callback, terminal_event_names=["error", "response.completed"])` What search retrieveStream() output looks like: the sample: <img width="1028" height="131" alt="image" src="https://github.com/user-attachments/assets/9b868d8d-0f45-4468-a2d7-bf02510e6905" /> the output: <img width="2886" height="705" alt="image" src="https://github.com/user-attachments/assets/5bca2fbf-34f2-4dda-b84a-f4d89148a036" /> --------- Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> Copilot-Session: e57edafe-9764-4b99-a1ae-0efd56e8729e Copilot-Session: 2a68a869-b075-46b2-87b4-2bbe62948e69 Copilot-Session: 51359cbf-bc53-429e-b947-8284a91d7d46
1 parent 10b1126 commit a0b0ee0

19 files changed

Lines changed: 2199 additions & 53 deletions

File tree

Lines changed: 17 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,17 @@
1+
---
2+
changeKind: feature
3+
packages:
4+
- "@typespec/http-client-python"
5+
---
6+
7+
Generate structured streaming client methods: operations whose HTTP response is a JSONL (`application/jsonl`) or SSE (`text/event-stream`) stream now return `Stream[T]` / `AsyncStream[T]`, yielding deserialized model instances instead of raw bytes.
8+
9+
The `Stream` / `AsyncStream` runtime (plus the JSONL / SSE decoders) is vendored at `_utils/streaming_base.py` and depends only on the released core runtime for the flavor — `azure.core.rest` for the Azure flavor and `corehttp.rest` for the unbranded flavor. These types are an internal implementation detail and are not part of the package's public API.
10+
11+
```python
12+
stream = client.receive()
13+
for thing in stream:
14+
...
15+
```
16+
17+
For SSE streams, the most recently received event `id` and `retry` value (if provided by the server) are exposed via `stream.last_event_id` / `stream.retry`. Event envelopes yield the `@Events.data` payload using its payload type and content type. JSON payload media types (`application/json` and `+json`) are decoded as JSON, while other SSE payload media types remain UTF-8 text for type-specific deserialization. Pass `last_event_id=` to an SSE operation to send `Last-Event-ID` when manually resuming a stream.

cspell.yaml

Lines changed: 7 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -5,13 +5,19 @@ dictionaries:
55
- node
66
- typescript
77
words:
8+
- aclose
9+
- aclosing
810
- Ablack
911
- Adoptium
12+
- aenter
13+
- aexit
1014
- agentic
1115
- agentics
1216
- aiohttp
17+
- aiter
1318
- alzimmer
1419
- amqp
20+
- anext
1521
- AQID
1622
- Arize
1723
- arizeaiobservabilityeval
@@ -121,6 +127,7 @@ words:
121127
- intrinsics
122128
- ints
123129
- IOHTTP
130+
- isascii
124131
- isdigit
125132
- isinstance
126133
- issecret

packages/http-client-python/emitter/src/http.ts

Lines changed: 154 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -18,6 +18,7 @@ import type {
1818
SdkQueryParameter,
1919
SdkServiceMethod,
2020
SdkServiceResponseHeader,
21+
SdkSseEventMetadata,
2122
SdkType,
2223
} from "@azure-tools/typespec-client-generator-core";
2324
import { getHttpOperationParameter, UsageFlags } from "@azure-tools/typespec-client-generator-core";
@@ -42,6 +43,158 @@ export enum ReferredByOperationTypes {
4243
NonPagingOnly = 2,
4344
}
4445

46+
type StructuredStreamKind = "jsonl" | "sse";
47+
type EmittedType = ReturnType<typeof getType>;
48+
49+
interface StructuredStreamEvent {
50+
eventType: string | undefined;
51+
/**
52+
* Payload type for this one SSE event. For an event envelope, this is the type of the property
53+
* marked `@Events.data`. Together with {@link eventType} these form the
54+
* runtime dispatch table (wire event name -> model to deserialize) inside the generated
55+
* `_callback`. This is a narrower type than {@link StructuredStreamingInfo.itemType}.
56+
*/
57+
payloadType: EmittedType;
58+
/**
59+
* True when this event is a `@terminalEvent` that carries a payload (a named / model event,
60+
* not a bare string-constant sentinel). Such events are deserialized and yielded like any
61+
* other event, and iteration stops immediately after one is yielded. Contrast with
62+
* {@link StructuredStreamingInfo.terminalEvent}, the sentinel that stops without yielding.
63+
*/
64+
isTerminal?: boolean;
65+
/** Content type of the payload, not the enclosing event. */
66+
payloadContentType?: string;
67+
}
68+
69+
interface StructuredStreamingInfo {
70+
kind: StructuredStreamKind;
71+
/**
72+
* The aggregate stream element type used for the `Stream[T]` / `AsyncStream[T]` return
73+
* annotation (a single type expression). For homogeneous JSONL this is the one model; for
74+
* heterogeneous SSE this is the union of every event payload.
75+
*
76+
* Note the deliberate overlap with the per-event {@link StructuredStreamEvent.payloadType}: for
77+
* heterogeneous SSE this union is exactly the sum of the `events[]` payload types. Both are
78+
* carried because the union alone cannot recover the wire-name -> member mapping needed for
79+
* dispatch, and the events list alone is not a single valid type expression for the annotation.
80+
*/
81+
itemType: EmittedType;
82+
events?: StructuredStreamEvent[];
83+
/**
84+
* A bare string-constant `@terminalEvent` with no event name (e.g. `"[DONE]"`). Iteration
85+
* stops when an event's `data` equals this value, and the sentinel is NOT yielded. Named /
86+
* model terminal events are carried in {@link events} with `isTerminal: true` instead.
87+
*/
88+
terminalEvent?: string;
89+
}
90+
91+
/** Whether pygen can deserialize the stream item type. */
92+
export function isStructuredStreamType(type: SdkType): boolean {
93+
switch (type.kind) {
94+
case "model":
95+
case "union":
96+
return true;
97+
case "nullable":
98+
return isStructuredStreamType(type.type);
99+
default:
100+
return false;
101+
}
102+
}
103+
104+
export function getStructuredStreamKind(
105+
response: SdkHttpResponse | SdkHttpErrorResponse,
106+
): StructuredStreamKind | undefined {
107+
if (response.sseMetadata) return "sse";
108+
109+
const contentTypes = response.streamMetadata?.contentTypes ?? response.contentTypes ?? [];
110+
for (const contentType of contentTypes) {
111+
const mediaType = contentType.split(";", 1)[0].trim().toLowerCase();
112+
if (mediaType === "text/event-stream") return "sse";
113+
if (mediaType === "application/jsonl") return "jsonl";
114+
}
115+
return undefined;
116+
}
117+
118+
function getStringConstantValue(type: SdkType): string | undefined {
119+
if (type.kind === "nullable") return getStringConstantValue(type.type);
120+
return type.kind === "constant" && typeof type.value === "string" ? type.value : undefined;
121+
}
122+
123+
/**
124+
* Split the SSE events into the runtime dispatch table and a bare string-constant sentinel.
125+
*
126+
* A `@terminalEvent` comes in two shapes:
127+
* * a nameless string constant (e.g. `"[DONE]"`) -> a pure sentinel: iteration stops when an
128+
* event's `data` equals this value and the event is NOT yielded. Returned as `terminalEvent`.
129+
* * a named / model event (e.g. `error`, `response.completed`) -> carries a payload the consumer
130+
* needs, so it is deserialized and yielded like any other event, then iteration stops. Returned
131+
* in `events` with `isTerminal: true`.
132+
*
133+
* `toPayloadType` maps event payloads to emitted types; it is injected so this partitioning stays
134+
* a pure function that can be unit-tested without a full emitter context.
135+
*/
136+
export function partitionSSEEvents(
137+
events: readonly SdkSseEventMetadata[],
138+
toPayloadType: (type: SdkType) => EmittedType,
139+
): { events: StructuredStreamEvent[]; terminalEvent?: string } {
140+
const dispatch: StructuredStreamEvent[] = [];
141+
let terminalEvent: string | undefined;
142+
for (const event of events) {
143+
if (event.isTerminalEvent) {
144+
const sentinelValue =
145+
event.eventType === undefined
146+
? (getStringConstantValue(event.payloadType) ?? getStringConstantValue(event.type))
147+
: undefined;
148+
if (sentinelValue !== undefined) {
149+
// Keep the first sentinel; no current spec defines more than one.
150+
terminalEvent ??= sentinelValue;
151+
continue;
152+
}
153+
dispatch.push({
154+
eventType: event.eventType,
155+
payloadType: toPayloadType(event.payloadType),
156+
isTerminal: true,
157+
payloadContentType: event.payloadContentType,
158+
});
159+
continue;
160+
}
161+
dispatch.push({
162+
eventType: event.eventType,
163+
payloadType: toPayloadType(event.payloadType),
164+
payloadContentType: event.payloadContentType,
165+
});
166+
}
167+
return terminalEvent !== undefined ? { events: dispatch, terminalEvent } : { events: dispatch };
168+
}
169+
170+
function emitStructuredStreamingInfo(
171+
context: PythonSdkContext,
172+
response: SdkHttpResponse | SdkHttpErrorResponse,
173+
): StructuredStreamingInfo | undefined {
174+
const streamMetadata = response.streamMetadata;
175+
if (!streamMetadata || !isStructuredStreamType(streamMetadata.streamType)) return undefined;
176+
177+
const kind = getStructuredStreamKind(response);
178+
if (!kind) return undefined;
179+
180+
const streaming: StructuredStreamingInfo = {
181+
kind,
182+
itemType: getType(context, streamMetadata.streamType),
183+
};
184+
if (kind !== "sse") return streaming;
185+
186+
const sseMetadata = response.sseMetadata;
187+
if (!sseMetadata || sseMetadata.events.length === 0) return undefined;
188+
189+
const { events, terminalEvent } = partitionSSEEvents(sseMetadata.events, (type) =>
190+
getType(context, type),
191+
);
192+
if (events.length > 0) streaming.events = events;
193+
if (terminalEvent !== undefined) streaming.terminalEvent = terminalEvent;
194+
195+
return streaming;
196+
}
197+
45198
function isEtagType(type: SdkType): boolean {
46199
if (type.kind === "nullable") return isEtagType(type.type);
47200
const raw = type.__raw;
@@ -682,6 +835,7 @@ function emitHttpResponse(
682835
"invalid-lro-result",
683836
method,
684837
),
838+
streaming: isException ? undefined : emitStructuredStreamingInfo(context, response),
685839
};
686840
}
687841

Lines changed: 155 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,155 @@
1+
import { strictEqual } from "assert";
2+
import { describe, it } from "vitest";
3+
import {
4+
getStructuredStreamKind,
5+
isStructuredStreamType,
6+
partitionSSEEvents,
7+
} from "../src/http.js";
8+
9+
describe("typespec-python: structured streaming", () => {
10+
it("treats model and union payloads as structured", () => {
11+
strictEqual(isStructuredStreamType({ kind: "model" } as any), true);
12+
strictEqual(isStructuredStreamType({ kind: "union" } as any), true);
13+
});
14+
15+
it("unwraps nullable payloads", () => {
16+
strictEqual(isStructuredStreamType({ kind: "nullable", type: { kind: "model" } } as any), true);
17+
strictEqual(
18+
isStructuredStreamType({ kind: "nullable", type: { kind: "bytes" } } as any),
19+
false,
20+
);
21+
});
22+
23+
it("treats bare byte/string payloads as unstructured", () => {
24+
strictEqual(isStructuredStreamType({ kind: "bytes" } as any), false);
25+
strictEqual(isStructuredStreamType({ kind: "string" } as any), false);
26+
});
27+
28+
it("detects the stream protocol explicitly", () => {
29+
strictEqual(getStructuredStreamKind({ sseMetadata: { events: [] } } as any), "sse");
30+
strictEqual(
31+
getStructuredStreamKind({
32+
streamMetadata: { contentTypes: ["text/event-stream; charset=utf-8"] },
33+
} as any),
34+
"sse",
35+
);
36+
strictEqual(
37+
getStructuredStreamKind({
38+
streamMetadata: { contentTypes: ["application/jsonl"] },
39+
} as any),
40+
"jsonl",
41+
);
42+
strictEqual(
43+
getStructuredStreamKind({
44+
streamMetadata: { contentTypes: ["application/json"] },
45+
} as any),
46+
undefined,
47+
);
48+
});
49+
50+
describe("terminal-event partitioning", () => {
51+
const identity = (payloadType: any) => payloadType;
52+
const model = (name: string) => ({ kind: "model", name });
53+
const constant = (value: string) => ({ kind: "constant", value });
54+
55+
it("keeps a nameless string-constant `[DONE]` as a drop-and-stop sentinel", () => {
56+
const created = model("ResponseCreated");
57+
const done = constant("[DONE]");
58+
const { events, terminalEvent } = partitionSSEEvents(
59+
[
60+
{
61+
eventType: "response.created",
62+
isTerminalEvent: false,
63+
type: created,
64+
payloadType: created,
65+
},
66+
{ eventType: undefined, isTerminalEvent: true, type: done, payloadType: done },
67+
] as any,
68+
identity,
69+
);
70+
// The sentinel is NOT a dispatch event; it only sets `terminalEvent`.
71+
strictEqual(terminalEvent, "[DONE]");
72+
strictEqual(events.length, 1);
73+
strictEqual(events[0].eventType, "response.created");
74+
strictEqual(events[0].isTerminal, undefined);
75+
});
76+
77+
it("keeps named / model `@terminalEvent`s in the dispatch table as yield-and-stop events", () => {
78+
const created = model("ResponseCreated");
79+
const completed = model("ResponseCompleted");
80+
const errored = model("StreamError");
81+
const { events, terminalEvent } = partitionSSEEvents(
82+
[
83+
{
84+
eventType: "response.created",
85+
isTerminalEvent: false,
86+
type: created,
87+
payloadType: created,
88+
},
89+
{
90+
eventType: "response.completed",
91+
isTerminalEvent: true,
92+
type: completed,
93+
payloadType: completed,
94+
},
95+
{ eventType: "error", isTerminalEvent: true, type: errored, payloadType: errored },
96+
] as any,
97+
identity,
98+
);
99+
// No bare sentinel: the two terminals carry payloads, so they stay in `events`.
100+
strictEqual(terminalEvent, undefined);
101+
strictEqual(events.length, 3);
102+
strictEqual(events[0].isTerminal, undefined);
103+
strictEqual(events[1].eventType, "response.completed");
104+
strictEqual(events[1].isTerminal, true);
105+
strictEqual(events[1].payloadType, completed);
106+
strictEqual(events[2].eventType, "error");
107+
strictEqual(events[2].isTerminal, true);
108+
strictEqual(events[2].payloadType, errored);
109+
});
110+
111+
it("supports a sentinel and named terminals together", () => {
112+
const delta = model("ResponseDelta");
113+
const completed = model("ResponseCompleted");
114+
const done = constant("[DONE]");
115+
const { events, terminalEvent } = partitionSSEEvents(
116+
[
117+
{ eventType: "response.delta", isTerminalEvent: false, type: delta, payloadType: delta },
118+
{
119+
eventType: "response.completed",
120+
isTerminalEvent: true,
121+
type: completed,
122+
payloadType: completed,
123+
},
124+
{ eventType: undefined, isTerminalEvent: true, type: done, payloadType: done },
125+
] as any,
126+
identity,
127+
);
128+
strictEqual(terminalEvent, "[DONE]");
129+
strictEqual(events.length, 2);
130+
strictEqual(events[0].isTerminal, undefined);
131+
strictEqual(events[1].eventType, "response.completed");
132+
strictEqual(events[1].isTerminal, true);
133+
});
134+
135+
it("emits the payload metadata for event envelopes", () => {
136+
const envelope = model("Envelope");
137+
const payload = { kind: "string" };
138+
const { events } = partitionSSEEvents(
139+
[
140+
{
141+
eventType: "withEnvelope",
142+
isTerminalEvent: false,
143+
type: envelope,
144+
payloadType: payload,
145+
isEventEnvelope: true,
146+
payloadContentType: "text/plain",
147+
},
148+
] as any,
149+
identity,
150+
);
151+
strictEqual(events[0].payloadType, payload);
152+
strictEqual(events[0].payloadContentType, "text/plain");
153+
});
154+
});
155+
});

0 commit comments

Comments
 (0)