feat(http-client-python): generate structured JSONL/SSE streaming

.chronus/changes/structured-streaming-2026-0-0.md

@@ -0,0 +1,17 @@
+---
+changeKind: feature
+packages:
+  - "@typespec/http-client-python"
+---
+
+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.
+
+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.
+
+```python
+stream = client.receive()
+for thing in stream:
+    ...
+```
+
+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

@@ -5,13 +5,19 @@ dictionaries:
   - node
   - typescript
 words:
+  - aclose
+  - aclosing
   - Ablack
   - Adoptium
+  - aenter
+  - aexit
   - agentic
   - agentics
   - aiohttp
+  - aiter
   - alzimmer
   - amqp
+  - anext
   - AQID
   - Arize
   - arizeaiobservabilityeval
@@ -121,6 +127,7 @@ words:
   - intrinsics
   - ints
   - IOHTTP
+  - isascii
   - isdigit
   - isinstance
   - issecret

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

@@ -18,6 +18,7 @@ import type {
   SdkQueryParameter,
   SdkServiceMethod,
   SdkServiceResponseHeader,
+  SdkSseEventMetadata,
   SdkType,
 } from "@azure-tools/typespec-client-generator-core";
 import { getHttpOperationParameter, UsageFlags } from "@azure-tools/typespec-client-generator-core";
@@ -42,6 +43,158 @@ export enum ReferredByOperationTypes {
   NonPagingOnly = 2,
 }
 
+type StructuredStreamKind = "jsonl" | "sse";
+type EmittedType = ReturnType<typeof getType>;
+
+interface StructuredStreamEvent {
+  eventType: string | undefined;
+  /**
+   * Payload type for this one SSE event. For an event envelope, this is the type of the property
+   * marked `@Events.data`. Together with {@link eventType} these form the
+   * runtime dispatch table (wire event name -> model to deserialize) inside the generated
+   * `_callback`. This is a narrower type than {@link StructuredStreamingInfo.itemType}.
+   */
+  payloadType: EmittedType;
+  /**
+   * True when this event is a `@terminalEvent` that carries a payload (a named / model event,
+   * not a bare string-constant sentinel). Such events are deserialized and yielded like any
+   * other event, and iteration stops immediately after one is yielded. Contrast with
+   * {@link StructuredStreamingInfo.terminalEvent}, the sentinel that stops without yielding.
+   */
+  isTerminal?: boolean;
+  /** Content type of the payload, not the enclosing event. */
+  payloadContentType?: string;
+}
+
+interface StructuredStreamingInfo {
+  kind: StructuredStreamKind;
+  /**
+   * The aggregate stream element type used for the `Stream[T]` / `AsyncStream[T]` return
+   * annotation (a single type expression). For homogeneous JSONL this is the one model; for
+   * heterogeneous SSE this is the union of every event payload.
+   *
+   * Note the deliberate overlap with the per-event {@link StructuredStreamEvent.payloadType}: for
+   * heterogeneous SSE this union is exactly the sum of the `events[]` payload types. Both are
+   * carried because the union alone cannot recover the wire-name -> member mapping needed for
+   * dispatch, and the events list alone is not a single valid type expression for the annotation.
+   */
+  itemType: EmittedType;
+  events?: StructuredStreamEvent[];
+  /**
+   * A bare string-constant `@terminalEvent` with no event name (e.g. `"[DONE]"`). Iteration
+   * stops when an event's `data` equals this value, and the sentinel is NOT yielded. Named /
+   * model terminal events are carried in {@link events} with `isTerminal: true` instead.
+   */
+  terminalEvent?: string;
+}
+
+/** Whether pygen can deserialize the stream item type. */
+export function isStructuredStreamType(type: SdkType): boolean {
+  switch (type.kind) {
+    case "model":
+    case "union":
+      return true;
+    case "nullable":
+      return isStructuredStreamType(type.type);
+    default:
+      return false;
+  }
+}
+
+export function getStructuredStreamKind(
+  response: SdkHttpResponse | SdkHttpErrorResponse,
+): StructuredStreamKind | undefined {
+  if (response.sseMetadata) return "sse";
+
+  const contentTypes = response.streamMetadata?.contentTypes ?? response.contentTypes ?? [];
+  for (const contentType of contentTypes) {
+    const mediaType = contentType.split(";", 1)[0].trim().toLowerCase();
+    if (mediaType === "text/event-stream") return "sse";
+    if (mediaType === "application/jsonl") return "jsonl";
+  }
+  return undefined;
+}
+
+function getStringConstantValue(type: SdkType): string | undefined {
+  if (type.kind === "nullable") return getStringConstantValue(type.type);
+  return type.kind === "constant" && typeof type.value === "string" ? type.value : undefined;
+}
+
+/**
+ * Split the SSE events into the runtime dispatch table and a bare string-constant sentinel.
+ *
+ * A `@terminalEvent` comes in two shapes:
+ *   * a nameless string constant (e.g. `"[DONE]"`) -> a pure sentinel: iteration stops when an
+ *     event's `data` equals this value and the event is NOT yielded. Returned as `terminalEvent`.
+ *   * a named / model event (e.g. `error`, `response.completed`) -> carries a payload the consumer
+ *     needs, so it is deserialized and yielded like any other event, then iteration stops. Returned
+ *     in `events` with `isTerminal: true`.
+ *
+ * `toPayloadType` maps event payloads to emitted types; it is injected so this partitioning stays
+ * a pure function that can be unit-tested without a full emitter context.
+ */
+export function partitionSSEEvents(
+  events: readonly SdkSseEventMetadata[],
+  toPayloadType: (type: SdkType) => EmittedType,
+): { events: StructuredStreamEvent[]; terminalEvent?: string } {
+  const dispatch: StructuredStreamEvent[] = [];
+  let terminalEvent: string | undefined;
+  for (const event of events) {
+    if (event.isTerminalEvent) {
+      const sentinelValue =
+        event.eventType === undefined
+          ? (getStringConstantValue(event.payloadType) ?? getStringConstantValue(event.type))
+          : undefined;
+      if (sentinelValue !== undefined) {
+        // Keep the first sentinel; no current spec defines more than one.
+        terminalEvent ??= sentinelValue;
+        continue;
+      }
+      dispatch.push({
+        eventType: event.eventType,
+        payloadType: toPayloadType(event.payloadType),
+        isTerminal: true,
+        payloadContentType: event.payloadContentType,
+      });
+      continue;
+    }
+    dispatch.push({
+      eventType: event.eventType,
+      payloadType: toPayloadType(event.payloadType),
+      payloadContentType: event.payloadContentType,
+    });
+  }
+  return terminalEvent !== undefined ? { events: dispatch, terminalEvent } : { events: dispatch };
+}
+
+function emitStructuredStreamingInfo(
+  context: PythonSdkContext,
+  response: SdkHttpResponse | SdkHttpErrorResponse,
+): StructuredStreamingInfo | undefined {
+  const streamMetadata = response.streamMetadata;
+  if (!streamMetadata || !isStructuredStreamType(streamMetadata.streamType)) return undefined;
+
+  const kind = getStructuredStreamKind(response);
+  if (!kind) return undefined;
+
+  const streaming: StructuredStreamingInfo = {
+    kind,
+    itemType: getType(context, streamMetadata.streamType),
+  };
+  if (kind !== "sse") return streaming;
+
+  const sseMetadata = response.sseMetadata;
+  if (!sseMetadata || sseMetadata.events.length === 0) return undefined;
+
+  const { events, terminalEvent } = partitionSSEEvents(sseMetadata.events, (type) =>
+    getType(context, type),
+  );
+  if (events.length > 0) streaming.events = events;
+  if (terminalEvent !== undefined) streaming.terminalEvent = terminalEvent;
+
+  return streaming;
+}
+
 function isEtagType(type: SdkType): boolean {
   if (type.kind === "nullable") return isEtagType(type.type);
   const raw = type.__raw;
@@ -682,6 +835,7 @@ function emitHttpResponse(
       "invalid-lro-result",
       method,
     ),
+    streaming: isException ? undefined : emitStructuredStreamingInfo(context, response),
   };
 }
 

packages/http-client-python/emitter/test/streaming.test.ts

@@ -0,0 +1,155 @@
+import { strictEqual } from "assert";
+import { describe, it } from "vitest";
+import {
+  getStructuredStreamKind,
+  isStructuredStreamType,
+  partitionSSEEvents,
+} from "../src/http.js";
+
+describe("typespec-python: structured streaming", () => {
+  it("treats model and union payloads as structured", () => {
+    strictEqual(isStructuredStreamType({ kind: "model" } as any), true);
+    strictEqual(isStructuredStreamType({ kind: "union" } as any), true);
+  });
+
+  it("unwraps nullable payloads", () => {
+    strictEqual(isStructuredStreamType({ kind: "nullable", type: { kind: "model" } } as any), true);
+    strictEqual(
+      isStructuredStreamType({ kind: "nullable", type: { kind: "bytes" } } as any),
+      false,
+    );
+  });
+
+  it("treats bare byte/string payloads as unstructured", () => {
+    strictEqual(isStructuredStreamType({ kind: "bytes" } as any), false);
+    strictEqual(isStructuredStreamType({ kind: "string" } as any), false);
+  });
+
+  it("detects the stream protocol explicitly", () => {
+    strictEqual(getStructuredStreamKind({ sseMetadata: { events: [] } } as any), "sse");
+    strictEqual(
+      getStructuredStreamKind({
+        streamMetadata: { contentTypes: ["text/event-stream; charset=utf-8"] },
+      } as any),
+      "sse",
+    );
+    strictEqual(
+      getStructuredStreamKind({
+        streamMetadata: { contentTypes: ["application/jsonl"] },
+      } as any),
+      "jsonl",
+    );
+    strictEqual(
+      getStructuredStreamKind({
+        streamMetadata: { contentTypes: ["application/json"] },
+      } as any),
+      undefined,
+    );
+  });
+
+  describe("terminal-event partitioning", () => {
+    const identity = (payloadType: any) => payloadType;
+    const model = (name: string) => ({ kind: "model", name });
+    const constant = (value: string) => ({ kind: "constant", value });
+
+    it("keeps a nameless string-constant `[DONE]` as a drop-and-stop sentinel", () => {
+      const created = model("ResponseCreated");
+      const done = constant("[DONE]");
+      const { events, terminalEvent } = partitionSSEEvents(
+        [
+          {
+            eventType: "response.created",
+            isTerminalEvent: false,
+            type: created,
+            payloadType: created,
+          },
+          { eventType: undefined, isTerminalEvent: true, type: done, payloadType: done },
+        ] as any,
+        identity,
+      );
+      // The sentinel is NOT a dispatch event; it only sets `terminalEvent`.
+      strictEqual(terminalEvent, "[DONE]");
+      strictEqual(events.length, 1);
+      strictEqual(events[0].eventType, "response.created");
+      strictEqual(events[0].isTerminal, undefined);
+    });
+
+    it("keeps named / model `@terminalEvent`s in the dispatch table as yield-and-stop events", () => {
+      const created = model("ResponseCreated");
+      const completed = model("ResponseCompleted");
+      const errored = model("StreamError");
+      const { events, terminalEvent } = partitionSSEEvents(
+        [
+          {
+            eventType: "response.created",
+            isTerminalEvent: false,
+            type: created,
+            payloadType: created,
+          },
+          {
+            eventType: "response.completed",
+            isTerminalEvent: true,
+            type: completed,
+            payloadType: completed,
+          },
+          { eventType: "error", isTerminalEvent: true, type: errored, payloadType: errored },
+        ] as any,
+        identity,
+      );
+      // No bare sentinel: the two terminals carry payloads, so they stay in `events`.
+      strictEqual(terminalEvent, undefined);
+      strictEqual(events.length, 3);
+      strictEqual(events[0].isTerminal, undefined);
+      strictEqual(events[1].eventType, "response.completed");
+      strictEqual(events[1].isTerminal, true);
+      strictEqual(events[1].payloadType, completed);
+      strictEqual(events[2].eventType, "error");
+      strictEqual(events[2].isTerminal, true);
+      strictEqual(events[2].payloadType, errored);
+    });
+
+    it("supports a sentinel and named terminals together", () => {
+      const delta = model("ResponseDelta");
+      const completed = model("ResponseCompleted");
+      const done = constant("[DONE]");
+      const { events, terminalEvent } = partitionSSEEvents(
+        [
+          { eventType: "response.delta", isTerminalEvent: false, type: delta, payloadType: delta },
+          {
+            eventType: "response.completed",
+            isTerminalEvent: true,
+            type: completed,
+            payloadType: completed,
+          },
+          { eventType: undefined, isTerminalEvent: true, type: done, payloadType: done },
+        ] as any,
+        identity,
+      );
+      strictEqual(terminalEvent, "[DONE]");
+      strictEqual(events.length, 2);
+      strictEqual(events[0].isTerminal, undefined);
+      strictEqual(events[1].eventType, "response.completed");
+      strictEqual(events[1].isTerminal, true);
+    });
+
+    it("emits the payload metadata for event envelopes", () => {
+      const envelope = model("Envelope");
+      const payload = { kind: "string" };
+      const { events } = partitionSSEEvents(
+        [
+          {
+            eventType: "withEnvelope",
+            isTerminalEvent: false,
+            type: envelope,
+            payloadType: payload,
+            isEventEnvelope: true,
+            payloadContentType: "text/plain",
+          },
+        ] as any,
+        identity,
+      );
+      strictEqual(events[0].payloadType, payload);
+      strictEqual(events[0].payloadContentType, "text/plain");
+    });
+  });
+});