feat(http-client-python): generate structured JSONL and SSE streams

Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com>

Copilot-Session: e57edafe-9764-4b99-a1ae-0efd56e8729e

Author: l0lawrence

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

@@ -0,0 +1,17 @@
+---
+changeKind: feature
+packages:
+  - "@typespec/http-client-python"
+---
+
+Generate structured streaming client methods for the **Azure flavor**: 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 into the generated package at `_utils/streaming_base.py` (like `_utils/model_base.py`), so it depends only on the released `azure.core.rest` — not on an unreleased `azure.core.streaming`.
+
+```python
+# For an operation returning JsonlStream<Thing> (Azure flavor):
+stream = client.receive()          # -> Stream[Thing]
+for thing in stream:               # deserialized model instances
+    ...
+```

cspell.yaml

@@ -7,11 +7,15 @@ dictionaries:
 words:
   - Ablack
   - Adoptium
+  - aenter
+  - aexit
   - agentic
   - agentics
   - aiohttp
+  - aiter
   - alzimmer
   - amqp
+  - anext
   - AQID
   - Arize
   - arizeaiobservabilityeval
@@ -120,6 +124,7 @@ words:
   - intrinsics
   - ints
   - IOHTTP
+  - isascii
   - isdigit
   - isinstance
   - issecret

packages/http-client-python/README.md

@@ -153,3 +153,20 @@ 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)
+
+For the **Azure flavor**, 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. For the unbranded flavor, streaming responses keep the existing raw byte-iterator behavior (`Iterator[bytes]` / `AsyncIterator[bytes]`).
+
+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.
+
+```python
+# For an operation returning JsonlStream<Thing> (Azure flavor):
+stream = client.receive()          # -> Stream[Thing]
+for thing in stream:               # deserialized model instances
+    ...
+```
+
+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 `azure.core.rest`, 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,88 @@ export enum ReferredByOperationTypes {
   NonPagingOnly = 2,
 }
 
+/**
+ * Determine whether a stream's payload type is "structured" (a model or union
+ * that can be deserialized into an item `T`), as opposed to a bare byte/string
+ * stream that should keep the existing raw byte-iterator behavior.
+ */
+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,
+): "jsonl" | "sse" | 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;
+}
+
+/**
+ * Build the `streaming` block for a response YAML when the response is a JSONL/SSE
+ * stream with a structured payload type (driven by the TCGC stream metadata).
+ *
+ * Returns `undefined` when structured streaming should not apply, in which case
+ * the existing raw byte-iterator behavior is preserved.
+ *
+ * For SSE, TCGC `sseMetadata` supplies each event's wire name, payload type, and
+ * terminal marker. JSONL only needs the common stream item type.
+ */
+function getStreamingInfo(
+  context: PythonSdkContext,
+  response: SdkHttpResponse | SdkHttpErrorResponse,
+): Record<string, any> | undefined {
+  const streamMetadata = response.streamMetadata;
+  if (!streamMetadata) return undefined;
+  if (!isStructuredStreamType(streamMetadata.streamType)) return undefined;
+  const kind = getStructuredStreamKind(response);
+  if (!kind) return undefined;
+
+  const streaming: Record<string, any> = {
+    kind,
+    itemType: getType(context, streamMetadata.streamType),
+  };
+
+  if (kind === "sse" && response.sseMetadata) {
+    const events: Record<string, any>[] = [];
+    let terminalEvent: string | undefined;
+    for (const event of response.sseMetadata.events) {
+      if (event.isTerminalEvent) {
+        terminalEvent =
+          getStringConstantValue(event.payloadType) ?? getStringConstantValue(event.type);
+      } else {
+        events.push({
+          eventType: event.eventType,
+          itemType: getType(context, event.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 +764,7 @@ function emitHttpResponse(
       "invalid-lro-result",
       method,
     ),
+    streaming: isException ? undefined : getStreamingInfo(context, response),
   };
 }
 

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

@@ -0,0 +1,45 @@
+import { strictEqual } from "assert";
+import { describe, it } from "vitest";
+import { getStructuredStreamKind, isStructuredStreamType } 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,
+    );
+  });
+});

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

@@ -279,12 +279,29 @@ 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.need_streaming_base
         )
 
     @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
+        )
+
+    @property
+    def need_streaming_base(self) -> bool:
+        # Whether to emit the vendored ``_utils/streaming_base.py`` (Stream / AsyncStream
+        # + JSONL / SSE decoders). Only needed when at least one operation returns a
+        # structured stream.
+        return self.has_structured_stream
+
     def need_utils_utils(self, async_mode: bool, client_namespace: str) -> bool:
         return (
             self.need_utils_form_data(async_mode, client_namespace)

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

@@ -98,12 +98,21 @@ def exact_name_params(self) -> set[str]:
 
     @property
     def stream_value(self) -> Union[str, bool]:
+        # Structured streams (JSONL / SSE) must always run the pipeline with
+        # stream=True so the body can be consumed incrementally by Stream/AsyncStream.
+        if self.has_structured_stream_response:
+            return True
         return (
             f'kwargs.pop("stream", {self.has_stream_response})'
             if self.expose_stream_keyword and self.has_response_body and "stream" not in self.exact_name_params
             else self.has_stream_response
         )
 
+    @property
+    def has_structured_stream_response(self) -> bool:
+        """Whether any success response is a structured (JSONL / SSE) stream returning Stream[T]."""
+        return any(getattr(r, "is_structured_stream", False) for r in self.responses)
+
     @property
     def has_form_data_body(self):
         return self.parameters.has_form_data_body