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

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

@@ -0,0 +1,15 @@
+---
+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:
+    ...
+```

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
@@ -120,6 +126,7 @@ words:
   - intrinsics
   - ints
   - IOHTTP
+  - isascii
   - isdigit
   - isinstance
   - issecret

packages/http-client-python/README.md

@@ -153,3 +153,24 @@ Whether to clear the output folder before generating the code. Defaults to `fals
 **Type:** `boolean`
 
 Emit YAML code model only, without running Python generator. For batch processing.
+
+## Structured streaming (JSONL / SSE)
+
+Operations whose HTTP response is a JSONL (`application/jsonl`) or SSE (`text/event-stream`) stream generate client methods that return `Stream[T]` (sync) / `AsyncStream[T]` (async), yielding deserialized model instances instead of raw bytes. This is driven by the TCGC response stream metadata (the response stream type) — there is no opt-in emitter option. This applies to both the Azure and unbranded flavors.
+
+For an operation returning `JsonlStream<Thing>`, the generated method returns `Stream[Thing]` (sync) / `AsyncStream[Thing]` (async), yielding deserialized `Thing` instances as each JSONL line arrives. Similarly, `SSEStream<Events>` produces a `Stream` / `AsyncStream` over the SSE event payloads.
+
+Generated packages re-export `Stream` and `AsyncStream` from their base namespace:
+
+```python
+from your_sdk import Stream
+from your_sdk.models import Thing
+
+stream: Stream[Thing] = client.receive()
+for thing in stream:
+    ...
+```
+
+The `Stream` / `AsyncStream` runtime (plus the JSONL and SSE decoders) is **vendored** into the generated package at `_utils/streaming_base.py` (alongside `_utils/model_base.py`). It depends only on the released core runtime for the flavor — `azure.core.rest` for the Azure flavor and `corehttp.rest` for the unbranded flavor — so no unreleased `azure.core.streaming` dependency is required at runtime.
+
+SSE `@events` unions use TCGC event metadata to deserialize each named event into its corresponding generated model. Events marked with `@terminalEvent` stop iteration without being yielded.

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

@@ -42,6 +42,157 @@ 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. 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}.
+   */
+  itemType: 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;
+}
+
+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.itemType}: 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;
+}
+
+/** The subset of an {@link SdkSseMetadata} event that terminal-event partitioning depends on. */
+interface SseEventLike {
+  eventType?: string | undefined;
+  isTerminalEvent: boolean;
+  type: SdkType;
+  payloadType: SdkType;
+}
+
+/**
+ * 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`.
+ *
+ * `toItemType` maps an event payload to its emitted type; 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 SseEventLike[],
+  toItemType: (payloadType: 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,
+        itemType: toItemType(event.payloadType),
+        isTerminal: true,
+      });
+      continue;
+    }
+    dispatch.push({ eventType: event.eventType, itemType: toItemType(event.payloadType) });
+  }
+  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),
+  };
+
+  const sseMetadata = response.sseMetadata;
+  if (!sseMetadata) return streaming;
+
+  const { events, terminalEvent } = partitionSseEvents(sseMetadata.events, (payloadType) =>
+    getType(context, payloadType),
+  );
+  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 +833,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,135 @@
+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].itemType, completed);
+      strictEqual(events[2].eventType, "error");
+      strictEqual(events[2].isTerminal, true);
+      strictEqual(events[2].itemType, 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);
+    });
+  });
+});

packages/http-client-python/generator/pygen/codegen/models/code_model.py

@@ -279,12 +279,22 @@ def need_utils_folder(self, async_mode: bool, client_namespace: str) -> bool:
             self.need_utils_utils(async_mode, client_namespace)
             or self.need_utils_serialization
             or self.options["models-mode"] == "dpg"
+            or self.has_structured_stream
         )
 
     @property
     def need_utils_serialization(self) -> bool:
         return not self.options["client-side-validation"]
 
+    @property
+    def has_structured_stream(self) -> bool:
+        return any(
+            op.has_structured_stream_response
+            for client in self.clients
+            for og in client.operation_groups
+            for op in og.operations
+        )
+
     def need_utils_utils(self, async_mode: bool, client_namespace: str) -> bool:
         return (
             self.need_utils_form_data(async_mode, client_namespace)