test integrating stream into the httpresponse

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

@@ -0,0 +1,12 @@
+---
+changeKind: feature
+packages:
+  - "@typespec/http-client-python"
+---
+
+Generate structured JSONL (`application/jsonl`) and SSE (`text/event-stream`) response streams for the Azure flavor. Generated methods return standard `Generator[T, None, None]` or `AsyncGenerator[T, None]` values that yield deserialized model instances. Unbranded response streams remain raw-byte iterators.
+
+```python
+for thing in client.receive():
+    ...
+```

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,30 @@ 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 `Generator[T, None, None]` (sync) or `AsyncGenerator[T, None]` (async). The generators yield deserialized model instances as each record arrives. This behavior is driven by TCGC response stream metadata; there is no emitter option. The unbranded flavor retains its existing raw-byte response behavior (`Iterator[bytes]` / `AsyncIterator[bytes]`).
+
+Fully consuming a generator closes the underlying response:
+
+```python
+for thing in client.receive():
+    ...
+```
+
+When stopping early, explicitly close the generator so the response is released:
+
+```python
+from contextlib import aclosing, closing
+
+with closing(client.receive()) as stream:
+    first = next(stream)
+
+async with aclosing(async_client.receive()) as stream:
+    first = await anext(stream)
+```
+
+The generated package keeps transport-neutral framing helpers private in `_utils/streaming.py`; generated operations own JSON parsing and model deserialization. The runtime uses only released `azure-core` APIs.
+
+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,89 @@ export enum ReferredByOperationTypes {
   NonPagingOnly = 2,
 }
 
+type StructuredStreamKind = "jsonl" | "sse";
+type EmittedType = ReturnType<typeof getType>;
+
+interface StructuredStreamEvent {
+  eventType: string | undefined;
+  itemType: EmittedType;
+}
+
+interface StructuredStreamingInfo {
+  kind: StructuredStreamKind;
+  itemType: EmittedType;
+  events?: StructuredStreamEvent[];
+  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 {
+  const contentTypes = response.streamMetadata?.contentTypes ?? response.contentTypes ?? [];
+  for (const contentType of contentTypes) {
+    if (contentType === "text/event-stream") return "sse";
+    if (contentType === "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;
+}
+
+export function emitStructuredStreamingInfo(
+  context: PythonSdkContext,
+  response: SdkHttpResponse | SdkHttpErrorResponse,
+): StructuredStreamingInfo | undefined {
+  if ((context.emitContext.options as any).flavor !== "azure") return 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 = sseMetadata.events
+    .filter((event) => !event.isTerminalEvent)
+    .map((event) => ({
+      eventType: event.eventType,
+      itemType: getType(context, event.payloadType),
+    }));
+  if (events.length > 0) streaming.events = events;
+
+  const terminalEvent = sseMetadata.events.find((event) => event.isTerminalEvent);
+  if (terminalEvent) {
+    const terminalEventValue =
+      getStringConstantValue(terminalEvent.payloadType) ??
+      getStringConstantValue(terminalEvent.type);
+    if (terminalEventValue !== undefined) streaming.terminalEvent = terminalEventValue;
+  }
+
+  return streaming;
+}
+
 function isEtagType(type: SdkType): boolean {
   if (type.kind === "nullable") return isEtagType(type.type);
   const raw = type.__raw;
@@ -682,6 +765,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,159 @@
+import { deepStrictEqual, strictEqual } from "assert";
+import { describe, it } from "vitest";
+import {
+  emitStructuredStreamingInfo,
+  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("requires exact structured streaming media types", () => {
+    strictEqual(
+      getStructuredStreamKind({
+        streamMetadata: { contentTypes: ["text/event-stream"] },
+      } as any),
+      "sse",
+    );
+    strictEqual(
+      getStructuredStreamKind({
+        streamMetadata: { contentTypes: ["application/jsonl"] },
+      } as any),
+      "jsonl",
+    );
+    strictEqual(
+      getStructuredStreamKind({
+        sseMetadata: { events: [] },
+        streamMetadata: { contentTypes: ["text/event-stream; charset=utf-8"] },
+      } as any),
+      undefined,
+    );
+    strictEqual(
+      getStructuredStreamKind({
+        streamMetadata: { contentTypes: ["application/json"] },
+      } as any),
+      undefined,
+    );
+  });
+
+  it("emits Azure JSONL metadata for a structured stream", () => {
+    const itemType = { kind: "model", name: "Item", properties: [{}] };
+    const context = {
+      emitContext: { options: { flavor: "azure" } },
+      __typesMap: new Map([[itemType, { type: "model", name: "Item" }]]),
+      __simpleTypesMap: new Map(),
+    };
+
+    deepStrictEqual(
+      emitStructuredStreamingInfo(
+        context as any,
+        {
+          streamMetadata: {
+            contentTypes: ["application/jsonl"],
+            streamType: itemType,
+          },
+        } as any,
+      ),
+      {
+        kind: "jsonl",
+        itemType: { type: "model", name: "Item" },
+      },
+    );
+  });
+
+  it("emits SSE event and terminal metadata", () => {
+    const itemType = { kind: "model", name: "Events", properties: [{}] };
+    const connected = { kind: "model", name: "Connected", properties: [{}] };
+    const message = { kind: "model", name: "Message", properties: [{}] };
+    const terminal = {
+      kind: "constant",
+      value: "[DONE]",
+      valueType: { kind: "string" },
+    };
+    const emitted = new Map([
+      [itemType, { type: "combined", name: "Events" }],
+      [connected, { type: "model", name: "Connected" }],
+      [message, { type: "model", name: "Message" }],
+    ]);
+    const context = {
+      emitContext: { options: { flavor: "azure" } },
+      __typesMap: emitted,
+      __simpleTypesMap: new Map(),
+    };
+
+    deepStrictEqual(
+      emitStructuredStreamingInfo(
+        context as any,
+        {
+          streamMetadata: {
+            contentTypes: ["text/event-stream"],
+            streamType: itemType,
+          },
+          sseMetadata: {
+            events: [
+              { eventType: "connected", payloadType: connected },
+              { eventType: undefined, payloadType: message },
+              {
+                eventType: undefined,
+                payloadType: terminal,
+                type: terminal,
+                isTerminalEvent: true,
+              },
+            ],
+          },
+        } as any,
+      ),
+      {
+        kind: "sse",
+        itemType: { type: "combined", name: "Events" },
+        events: [
+          {
+            eventType: "connected",
+            itemType: { type: "model", name: "Connected" },
+          },
+          {
+            eventType: undefined,
+            itemType: { type: "model", name: "Message" },
+          },
+        ],
+        terminalEvent: "[DONE]",
+      },
+    );
+  });
+
+  it("explicitly excludes unbranded generation", () => {
+    strictEqual(
+      emitStructuredStreamingInfo(
+        {
+          emitContext: { options: { flavor: "unbranded" } },
+          __typesMap: new Map(),
+          __simpleTypesMap: new Map(),
+        } as any,
+        {
+          streamMetadata: {
+            contentTypes: ["application/jsonl"],
+            streamType: { kind: "model" },
+          },
+        } as any,
+      ),
+      undefined,
+    );
+  });
+});

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

@@ -279,12 +279,27 @@ 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
         )
 
     @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(self) -> bool:
+        """Whether to emit the private structured-stream framing helpers."""
+        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,20 @@ def exact_name_params(self) -> set[str]:
 
     @property
     def stream_value(self) -> Union[str, bool]:
+        # Structured streams must always run the pipeline incrementally.
+        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 or SSE stream."""
+        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